wehub-resource-sync
49b9bb6724
Deploy Docs / deploy-docs (push) Failing after 1s
Conformance Tests / client-conformance (push) Failing after 3s
Conformance Tests / server-conformance (push) Failing after 1s
GitHub Actions Security Analysis / zizmor (push) Failing after 1s
CI / checks (push) Failing after 59m20s
CI / all-green (push) Waiting to run
83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
"""cd to the `examples/snippets/clients` directory and run:
|
|
uv run client
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
|
|
import mcp_types as types
|
|
|
|
from mcp import ClientSession, StdioServerParameters
|
|
from mcp.client.context import ClientRequestContext
|
|
from mcp.client.stdio import stdio_client
|
|
|
|
# Create server parameters for stdio connection
|
|
server_params = StdioServerParameters(
|
|
command="uv", # Using uv to run the server
|
|
args=["run", "server", "mcpserver_quickstart", "stdio"], # We're already in snippets dir
|
|
env={"UV_INDEX": os.environ.get("UV_INDEX", "")},
|
|
)
|
|
|
|
|
|
# Optional: create a sampling callback
|
|
async def handle_sampling_message(
|
|
context: ClientRequestContext, params: types.CreateMessageRequestParams
|
|
) -> types.CreateMessageResult:
|
|
print(f"Sampling request: {params.messages}")
|
|
return types.CreateMessageResult(
|
|
role="assistant",
|
|
content=types.TextContent(
|
|
type="text",
|
|
text="Hello, world! from model",
|
|
),
|
|
model="gpt-3.5-turbo",
|
|
stop_reason="endTurn",
|
|
)
|
|
|
|
|
|
async def run():
|
|
async with stdio_client(server_params) as (read, write):
|
|
async with ClientSession(read, write, sampling_callback=handle_sampling_message) as session:
|
|
# Initialize the connection
|
|
await session.initialize()
|
|
|
|
# List available prompts
|
|
prompts = await session.list_prompts()
|
|
print(f"Available prompts: {[p.name for p in prompts.prompts]}")
|
|
|
|
# Get a prompt (greet_user prompt from mcpserver_quickstart)
|
|
if prompts.prompts:
|
|
prompt = await session.get_prompt("greet_user", arguments={"name": "Alice", "style": "friendly"})
|
|
print(f"Prompt result: {prompt.messages[0].content}")
|
|
|
|
# List available resources
|
|
resources = await session.list_resources()
|
|
print(f"Available resources: {[r.uri for r in resources.resources]}")
|
|
|
|
# List available tools
|
|
tools = await session.list_tools()
|
|
print(f"Available tools: {[t.name for t in tools.tools]}")
|
|
|
|
# Read a resource (greeting resource from mcpserver_quickstart)
|
|
resource_content = await session.read_resource("greeting://World")
|
|
content_block = resource_content.contents[0]
|
|
if isinstance(content_block, types.TextResourceContents):
|
|
print(f"Resource content: {content_block.text}")
|
|
|
|
# Call a tool (add tool from mcpserver_quickstart)
|
|
result = await session.call_tool("add", arguments={"a": 5, "b": 3})
|
|
result_unstructured = result.content[0]
|
|
if isinstance(result_unstructured, types.TextContent):
|
|
print(f"Tool result: {result_unstructured.text}")
|
|
result_structured = result.structured_content
|
|
print(f"Structured tool result: {result_structured}")
|
|
|
|
|
|
def main():
|
|
"""Entry point for the client script."""
|
|
asyncio.run(run())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|