Mmsp Python

uv add mmsp
# or
pip install mmsp

import asyncio
from mmsp import AutoLLMClient


def get_weather(location: str) -> str:
    return f"Temperature in {location}: 22 C"


# Map tool names to their implementations so calls can be dispatched by name.
TOOLS = {"get_weather": get_weather}


async def main():
    weather_function = {
        "name": "get_weather",
        "description": "Gets the current weather for a given location.",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "The city name"
                }
            },
            "required": ["location"]
        }
    }

    client = AutoLLMClient(model="gpt-5.5")
    config = {"tools": [weather_function]}

    tool_call = None
    async for event in client.streaming_response_stateful(
        message={
            "role": "user",
            "content_items": [{"type": "text.done", "text": "What's the weather in London?"}]
        },
        config=config
    ):
        if event["event_type"] == "stop":
            # Always the last event, exactly once: the response has finished.
            print(event["finish_reason"], event["usage_metadata"])
        for item in event["content_items"]:
            if item["type"] == "tool_call.done":  # the complete call; tool_call.delta items are fragments
                tool_call = item

    if tool_call:
        # Dispatch by tool name instead of hardcoding the function.
        result = TOOLS[tool_call["name"]](**tool_call["arguments"])

        async for event in client.streaming_response_stateful(
            message={
                "role": "user",
                "content_items": [
                    {
                        "type": "tool_result.done",
                        "text": result,
                        "tool_call_id": tool_call["tool_call_id"]
                    }
                ]
            },
            config=config
        ):
            print(event)
            # Streams the answer as text.delta fragments, closes it with text.done, then one stop event carrying usage:
            # {'role': 'assistant', 'event_type': 'delta', 'content_items': [{'type': 'text.delta', 'text': 'The'}], 'usage_metadata': None, 'finish_reason': None}
            # {'role': 'assistant', 'event_type': 'delta', 'content_items': [{'type': 'text.delta', 'text': ' weather'}], 'usage_metadata': None, 'finish_reason': None}
            # {'role': 'assistant', 'event_type': 'delta', 'content_items': [{'type': 'text.delta', 'text': ' is 22 C.'}], 'usage_metadata': None, 'finish_reason': None}
            # {'role': 'assistant', 'event_type': 'delta', 'content_items': [{'type': 'text.done', 'text': 'The weather is 22 C.'}], 'usage_metadata': None, 'finish_reason': None}
            # {'role': 'assistant', 'event_type': 'stop', 'content_items': [], 'usage_metadata': {'cached_tokens': 0, 'prompt_tokens': 12, 'thoughts_tokens': 0, 'response_tokens': 8}, 'finish_reason': 'stop'}


asyncio.run(main())