Skip to content

conversions

Class info

🛈 DocStrings

Conversions between internal and MCP types.

convert_mcp_content async 

convert_mcp_content(
    mcp_content: Sequence[ContentBlock | TextResourceContents | BlobResourceContents],
    client: Client[Any] | None = None,
) -> list[str | BinaryContent]

Convert MCP content blocks to PydanticAI content types.

If a FastMCP client is given, this function will try to resolve the ResourceLinks.

Source code in src/llmling_agent/mcp_server/conversions.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def convert_mcp_content(
    mcp_content: Sequence[ContentBlock | TextResourceContents | BlobResourceContents],
    client: Client[Any] | None = None,
) -> list[str | BinaryContent]:
    """Convert MCP content blocks to PydanticAI content types.

    If a FastMCP client is given, this function will try to resolve the ResourceLinks.

    """
    contents: list[Any] = []

    for block in mcp_content:
        match block:
            case TextContent(text=text):
                contents.append(text)
            case TextResourceContents(text=text):
                contents.append(text)
            case ImageContent(data=data, mimeType=mime_type):
                decoded_data = base64.b64decode(data)
                img = BinaryImage(data=decoded_data, media_type=mime_type)
                contents.append(img)
            case AudioContent(data=data, mimeType=mime_type):
                decoded_data = base64.b64decode(data)
                content = BinaryContent(data=decoded_data, media_type=mime_type)
                contents.append(content)
            case BlobResourceContents(blob=blob):
                decoded_data = base64.b64decode(blob)
                mime = "application/octet-stream"
                content = BinaryContent(data=decoded_data, media_type=mime)
                contents.append(content)
            case ResourceLink(uri=uri):
                if client:
                    try:
                        res = await client.read_resource(uri)
                        nested = await convert_mcp_content(res)
                        contents.extend(nested)
                    except Exception:  # noqa: BLE001
                        # Fallback to DocumentUrl if reading fails
                        logger.warning("Failed to read resource", uri=uri)
                contents.append(DocumentUrl(url=str(uri)))
            case EmbeddedResource(resource=TextResourceContents(text=text)):
                contents.append(text)
            case EmbeddedResource(resource=BlobResourceContents() as blob_resource):
                contents.append(f"[Binary data: {blob_resource.mimeType}]")
            case _:
                contents.append(str(block))  # Convert anything else to string
    return contents

to_mcp_messages 

to_mcp_messages(part: ModelRequestPart | ModelResponsePart) -> list[PromptMessage]

Convert internal PromptMessage to MCP PromptMessage.

Source code in src/llmling_agent/mcp_server/conversions.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def to_mcp_messages(
    part: ModelRequestPart | ModelResponsePart,
) -> list[PromptMessage]:
    """Convert internal PromptMessage to MCP PromptMessage."""
    messages = []
    match part:
        case UserPromptPart(content=str() as c):
            content = TextContent(type="text", text=c)
            messages.append(PromptMessage(role="user", content=content))
        case UserPromptPart(content=content_items):
            for item in content_items:
                match item:
                    case BinaryContent():
                        if item.is_audio:
                            encoded = base64.b64encode(item.data).decode("utf-8")
                            audio = AudioContent(
                                type="audio", data=encoded, mimeType=item.media_type
                            )
                            messages.append(PromptMessage(role="user", content=audio))
                        elif item.is_image:
                            encoded = base64.b64encode(item.data).decode("utf-8")
                            image = ImageContent(
                                type="image", data=encoded, mimeType=item.media_type
                            )
                        messages.append(PromptMessage(role="user", content=image))
                    case FileUrl(url=url):
                        content = TextContent(type="text", text=url)
                        messages.append(PromptMessage(role="user", content=content))

        case SystemPromptPart(content=msg):
            messages.append(PromptMessage(role="user", content=TextContent(type="text", text=msg)))
        case TextPart(content=msg):
            messages.append(
                PromptMessage(role="assistant", content=TextContent(type="text", text=msg))
            )
    return messages