forked from mcp-use/mcp-use
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdirect_tool_call.py
More file actions
59 lines (45 loc) · 1.92 KB
/
direct_tool_call.py
File metadata and controls
59 lines (45 loc) · 1.92 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
"""
Direct tool calling example using MCPClient.
Notes:
- This demonstrates calling tools directly without an LLM/agent.
- This approach will not work for tools that require sampling.
"""
import asyncio
from mcp_use import MCPClient
config = {
"mcpServers": {
"everything": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-everything"],
}
}
}
async def call_tool_example() -> None:
client = MCPClient.from_dict(config)
try:
# Create and initialize sessions for configured servers
await client.create_all_sessions()
# Retrieve the session for the "everything" server (match the server name key in the config)
session = client.get_session("everything")
# List available tools
tools = await session.list_tools()
tool_names = [t.name for t in tools]
print(f"Available tools: {tool_names}")
# Choose a tool to call (note: do not call sampling tools because they require an LLM to complete)
# In the example, we call the "add" tool from the "everything" server
# Result is a CallToolResult object
# - content is a list of ContentBlock objects
# - structuredContent is a dictionary of the structured result of the tool call (only for non-sampling tools)
# - isError is a boolean indicating if the tool call was successful
result_tool_add = await session.call_tool(name="add", arguments={"a": 1, "b": 2})
# Handle and print the result
if getattr(result_tool_add, "isError", False):
print(f"Error: {result_tool_add.content}")
else:
print(f"Tool result: {result_tool_add.content}")
print(f"Text result: {result_tool_add.content[0].text}")
finally:
# Ensure we clean up resources properly
await client.close_all_sessions()
if __name__ == "__main__":
asyncio.run(call_tool_example())