mcp_client.py

#!/usr/bin/env python3

import json

import anyio
from jsonschema import validate
from mcp import ReadResourceResult
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
from mcp.types import CallToolResult
from pydantic import BaseModel


class WeatherResult(BaseModel):
    location: str
    conditions: str

async def run():
    server_params = StdioServerParameters(command="python", args=["mcp_server.py"])

    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # Initialize the session
            await session.initialize()

            schema_resource: ReadResourceResult = await session.read_resource("my-mcp://schema/tools/check_weather/schema.json")
            schema =  json.loads(schema_resource.contents[0].text)

            ### LLM (Assitant) has stopped for tool_use
            tool_result: CallToolResult = await session.call_tool("check_weather", {"location": "London"})
            payload = tool_result.content[0].resource.text
            validate(instance=json.loads(payload),schema=schema)

            structured = WeatherResult.model_validate_json(payload)
            ### Send new User message to LLM (Tool Result)
            print(f"Weather in {structured.location}: {structured.conditions}")

if __name__ == "__main__":
    anyio.run(run)

mcp_server.py

#!/usr/bin/env python3
from string import Template

from mcp.server.fastmcp import FastMCP
from mcp.types import EmbeddedResource


WEATHER_SCHEMA = """
{
  "type": "object",
  "properties": {
    "location": {
      "type": "string",
      "description": "Name of the city or location"
    },
    "conditions": {
      "type": "string",
      "description": "Current weather conditions"
    }
  },
  "required": ["location", "conditions"]
}
"""

WEATHER_TEMPLATE = """
{
  "location": "$location",
  "conditions": "Sunny"
}
"""

# Create the FastMCP server
app = FastMCP(name="structured weather server")


@app.tool(name="check_weather", description="Gets the weather as JSON.")
def check_weather(location: str) -> list[EmbeddedResource]:
    result: str = Template(WEATHER_TEMPLATE).substitute(location=location)
    return EmbeddedResource(
        type="resource",
        resource={"mimeType":"application/json",
        "uri":f"my-mcp://check_weather/{location}",
        "text":result}
    )

@app.resource("my-mcp://schema/tools/check_weather/schema.json",
              mime_type="application/schema+json")
def get_schema() -> str:
    return WEATHER_SCHEMA

if __name__ == "__main__":
    app.run(transport="stdio")