Skip to main content
This example shows how to wrap an existing Pydantic AI translation agent with a MeshAgent TaskRunner. We subclass the base TaskRunner to create a TranslationTaskRunner with a defined input schema, output schema, and ask() method. This allows us to keep the existing logic from the Pydantic AI agent and run it inside a MeshAgent room. The sample uses the MeshAgent room router for Anthropic access, so credentials are supplied by your project configuration.

​
Prerequisites

  • Python venv with meshagent and pydantic_ai installed
  • Authenticate to MeshAgent meshagent setup

​
Step 1: Create the TaskRunner

Copy this code into a file called translator.py. This example uses the MeshAgent room router to supply Anthropic credentials, so you do not need a local ANTHROPIC_API_KEY. If you are not using the room router, update the provider to use your own Anthropic key or swap in a different model/provider. 
import json
import asyncio
import logging
import os
from typing import Optional
from datetime import date, datetime
from meshagent.otel import otel_config
from meshagent.api.services import ServiceHost
from meshagent.agents import TaskRunner

from pydantic_ai import Agent
from pydantic import BaseModel, Field, ConfigDict
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.providers.anthropic import AnthropicProvider

from meshagent.anthropic.proxy import get_client as get_anthropic_client

otel_config(service_name="translator")
log = logging.getLogger("translator")

service = ServiceHost()  # port defaults to an available port if not assigned

# Define Inputs, Outputs, and Pydantic AI Agent for Translation
class TranslationInput(BaseModel):
    text: str = Field(..., description="Text to translate")
    model_config = ConfigDict(extra="forbid")


class TranslationResult(BaseModel):
    french_translation: str
    spanish_translation: str
    model_config = ConfigDict(extra="forbid")


system_prompt = f"""
    # Role Background
    You are responsible for translating recent news announcements into other languages. You are exposed to a variety of topics beyond your knowledge cutoff date. The current date is: {date.today().strftime("%B %d, %Y")}

    # Task
    Provide two translations, one in French and one in Spanish.    
    """

def build_translation_agent(*, room):
    # Use the MeshAgent room proxy so API keys are provided by the room router.
    client = get_anthropic_client(room=room)
    provider = AnthropicProvider(anthropic_client=client)
    return Agent(
        model=AnthropicModel(model_name="claude-sonnet-4-5-20250929", provider=provider),
        deps_type=None,
        instructions=system_prompt,
        output_type=TranslationResult,
    )


# Utility function
async def save_to_storage(room, path: str, data: bytes):
    handle = await room.storage.open(path=path, overwrite=True)
    await room.storage.write(handle=handle, data=data)
    await room.storage.close(handle=handle)


# Use Pydantic AI agent in a MeshAgent Room
@service.path(path="/translator", identity="translator")
class TranslationTaskRunner(TaskRunner):
    def __init__(self):
        super().__init__(
            description="An agent that translates text to French and Spanish",
            input_schema=TranslationInput.model_json_schema(),
            output_schema=TranslationResult.model_json_schema(),
        )

    async def ask(self, *, context, arguments, attachment: Optional[bytes] = None):
        room = context.room

        inputs = TranslationInput(**arguments)
        log.info(f"Translating Text: {inputs.text}")

        translation_agent = build_translation_agent(room=room)
        translations = await translation_agent.run(inputs.text)
        log.info(f"Translation Result: {translations.output}")

        # save results to room storage
        log.info("Translation completed, writing raw results to Room storage.")

        await save_to_storage(
            room=room,
            path=f"translations/{room.room_name}-translation-{datetime.now():%Y%m%dT%H%M%S}.json",
            data=json.dumps(
                {
                    "input_text": inputs.text,
                    "translations": translations.output.model_dump(),
                },
                indent=2,
                ensure_ascii=False,
            ).encode("utf-8"),
        )

        return translations.output.model_dump()

asyncio.run(service.run())

​
Step 2: Run locally

From the terminal start your service locally. Be sure you are in an activated virtual environment where meshagent and pydantic_ai are installed:
bash
meshagent setup # this will authenticate you 
meshagent service run "translator.py" --room=translate

​
Step 3: Invoke and Test the TaskRunner

You can test the TaskRunner by invoking it from MeshAgent Studio, the CLI, or SDK code. If you want to make changes to the agent you can run Ctrl+C to stop the service from running, update it, then restart the service. If you want the service to persist you will need to deploy it (see Step 4).

​
Invoking from the Studio

  1. Go to MeshAgent Studio
  2. Enter the room translate
  3. Click the upper left menu —> “Toolkits”
  4. Select translator from the agent dropdown
  5. Enter the text to translate (you will ned to supply the JSON schema e.g. {"text":"How are you today?"})
  6. Results appear and are saved to room storage under the “translations” folder

​
Invoking from the CLI or Code

From the CLI While the service is running, create a new tab in your terminal and run:
CLI
meshagent room agents invoke-tool \
  --room=translate \
  --toolkit=translator \
  --tool=run_translator_task \
  --arguments='{"text":"I love MeshAgent!"}'
This will return a result like: 
Connecting to room...
Invoking tool...
{"french_translation": "J'adore MeshAgent !", "spanish_translation": "\u00a1Me encanta MeshAgent!"}
From Code Create and run a file called invoke_translator.py. This will establish the connection to the room and allow you to run the TranslationTaskRunner. 
import os
import asyncio
import json
import logging
from typing import Dict, Any
from meshagent.api import (
    RoomClient,
    WebSocketClientProtocol,
    ParticipantToken,
    ApiScope,
    ParticipantGrant,
)
from meshagent.api.helpers import websocket_room_url
from meshagent.otel import otel_config

otel_config()
log = logging.getLogger(__name__)

api_key = os.getenv("MESHAGENT_API_KEY")
if not api_key:
    raise RuntimeError("Set MESHAGENT_API_KEY before running this script.")


async def call_agent(
    room_name: str, agent_name: str, arguments: Dict[str, Any]
) -> Dict[str, Any]:
    """Invoke a TaskRunner tool with the given arguments."""
    token = ParticipantToken(
        name="sample-participant",
        grants=[
            ParticipantGrant(name="room", scope=room_name),
            ParticipantGrant(name="role", scope="agent"),
            ParticipantGrant(name="api", scope=ApiScope.agent_default()),
        ],
    ).to_jwt(api_key=api_key)

    protocol = WebSocketClientProtocol(
        url=websocket_room_url(room_name=room_name), token=token
    )
    try:
        async with RoomClient(protocol=protocol) as room:
            log.info(f"Connected to room: {room.room_name}")
            tool_name = f"run_{agent_name}_task"
            result = await room.agents.invoke_tool(
                toolkit=agent_name,
                tool=tool_name,
                arguments=arguments,
            )
            # Extract JSON data from JsonBody response
            return result.json if hasattr(result, "json") else result
    except Exception as e:
        log.error(f"Connection failed:{e}")
        raise


async def main():
    # Call your translator
    result = await call_agent(
        room_name="translate",
        agent_name="translator",
        arguments={"text": "Hello, how are you today?"},
    )

    print("Translation result:")
    print(json.dumps(result, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    asyncio.run(main())
Be sure your service is still running locally, then from a separate tab in your terminal run:
bash
# Create, Activate, and export your MESHAGENT_API_KEY if you haven't already
# meshagent api-key create my-api-key activate
# export MESHAGENT_API_KEY="xxxxxxx"
python invoke_translator.py
You will see the result logged to your terminal and saved in the room.

​
Step 4: Deploy as a service

To persist the agent you will need to deploy it to your room. This sample uses a single image so the container includes the pydantic_ai dependency. Dockerfile (single image) Build from the sample folder so the Dockerfile copies the translator files into the image. 
FROM meshagent/python-sdk:latest

WORKDIR /src
COPY . /src

RUN pip install --no-cache-dir pydantic-ai
To build and push the container run:
bash
docker buildx build . \
  -t "<REGISTRY>/<NAMESPACE>/pydanticai-translator:<TAG>" \
  --platform linux/amd64 \
  --push
Note: If you build your own image, we recommend optimizing it with eStargz for faster pulls.
meshagent.yaml Define the service configuration in a meshagent.yaml file. This service references your single image and runs the TaskRunner entrypoint. 
kind: Service
version: v1
metadata:
  name: pydanticai-translator
  description: "A TaskRunner that translates text with PydanticAI"
  annotations:
    meshagent.service.id: "pydanticai-translator"
agents:
  - name: translator
    description: "Translate text to French and Spanish"
    annotations:
      meshagent.agent.type: "TaskRunner"
ports:
- num: "*"
  endpoints:
  - path: /translator
    meshagent:
      identity: translator
container:
  # Build this image from the Dockerfile in this folder.
  image: "<REGISTRY>/<NAMESPACE>/pydanticai-translator:<TAG>"
  command: python /src/translator.py
Deploy it to your room Next from the CLI in the directory where your meshagent.yaml file is run:
bash
# Room service
meshagent service create --file meshagent.yaml --room translate
The agent is now deployed to the translate room! Now the agent will always be available inside the room for us to use. You can invoke it by following the same steps in step 3 above.

​
Next Steps