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.

​
Prerequisites

  • Python venv with meshagent and pydantic_ai installed
  • An LLM provider key (example uses ANTHROPIC_API_KEY)
  • Authenticate to MeshAgent meshagent setup

​
Example: Integrating a Pydantic AI Agent

Copy this code into a file called translator.py. You will need to create and export an ANTHROPIC_API_KEY first for this to run. Or update the example to use a model from a different provider of your choice. 
import json
import asyncio
import logging
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

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.    
    """
translation_agent = Agent(
    model=AnthropicModel(
        "claude-4-sonnet-20250514", provider=AnthropicProvider()
    ),  # pass API key from env variables
    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__(
            name="translator",
            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):
        room = context.room

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

        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())

​
Running the TaskRunner

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

​
Invoking from the Studio

  1. Go to MeshAgent Studio
  2. Enter the room translate
  3. Click the upper left menu —> “Run Task”
  4. Select translator from the agent dropdown
  5. Enter the text to translate
  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 agents ask \
  --room=translate \
  --agent=translator \
  --input='{"text":"I love MeshAgent!"}'
This will return a result like: 
Connecting to room...
Asking agent...
{"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]:
    """Call a MeshAgent agent 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}")
            result = await room.agents.ask(agent=agent_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.

​
Example: Using TaskRunner Agents as Agent Tools

TaskRunners can also be surfaced in toolkits and used as tools by other agents using RunTaskTool. Exposing agents as tools and then giving those tools to a ChatBot is a great way to get started building multi-agent systems. Let’s try this out by creating a new ChatBot and giving it a tool, the Pydantic AI translation agent that we defined above.

​
Creating the service with agents-as-tools

At the end of our translator.py file let’s add A new ChatBot service and pass it the translator agent as a RequiredToolkit. This will allow us to chat with the ChatBot while the ChatBot uses the translator agent as a tool when necessary.
Python
from meshagent.openai import OpenAIResponsesAdapter
from meshagent.api import RequiredToolkit
from meshagent.agents.chat import ChatBot

@service.path(path="/chatwithtools", identity="chat-translator")
class ChatBotAgentTools(ChatBot):
    def __init__(self):
        super().__init__(
            name="chat-translator",
            llm_adapter=OpenAIResponsesAdapter(),
            requires=[RequiredToolkit(name="agents", tools=["translator"])],
        )
The ServiceHost will automatically discover and start both the translation agent and the chatbot since they are part of the same service.
bash
meshagent service run "translator.py" --room=translate
Now you can interact with the chat agent from the studio and have it create translations for you! To use the chatbot:
  1. Go to studio.meshagent.com
  2. Enter the room translate
  3. In the participants list, you’ll see the chat-translator
  4. Click on chat-translator to start chatting with the agent
  5. Ask it something like: “Can you translate ‘Hello, how are you?’ for me?”
  6. The chat-translator will automatically use the translator Pydantic AI TranslationTaskRunner as a tool to complete the request and you will see the JSON results added to the translations folder in the room storage.
Remember: You can still use the Pydantic AI TranslationTaskRunner directly since it is running in the room, this approach just demonstrates an easy way to use a TaskRunner agent as a tool for another agent.

​
Next Steps

Dive Deeper into TaskRunners Learn how to deploy agents with MeshAgent