Skip to main content
MeshAgent already gives you logs, traces, metrics, session data, and developer logs in MeshAgent Studio. Use this page when you want to add your own spans, logs, and metrics inside a custom Python service. For the built-in observability model first, see Observability.

​
Enable telemetry in your service

Add two lines at startup before you create your ServiceHost:
Python
from meshagent.otel import otel_config
otel_config(service_name="weather-service")
When this code runs inside MeshAgent, otel_config() uses the injected OTEL_ENDPOINT and room/session environment to send telemetry to MeshAgent Studio.

​
Example: Weather Toolkit with custom instrumentation

The example adds custom spans around validation, the external HTTP call, and response parsing. It also adds logs and metrics. 
import asyncio
import httpx
import logging
from meshagent.api.services import ServiceHost
from meshagent.tools import FunctionTool, ToolContext, RemoteToolkit
from meshagent.otel import otel_config
from opentelemetry import trace, metrics
from opentelemetry.trace import Status, StatusCode

# Configure OpenTelemetry
otel_config(service_name="weather_tools")
service = ServiceHost()
log = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

# Counters
calls = meter.create_counter(
    "weather.calls", unit="1", description="Total weather tool invocations"
)
errors = meter.create_counter(
    "weather.errors", unit="1", description="Errors during execution"
)


class WeatherTool(FunctionTool):
    def __init__(self):
        super().__init__(
            name="get_weather",
            title="Weather Tool",
            description="Get current weather for a city using wttr.in API",
            input_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["city", "units"],
                "properties": {
                    "city": {"type": "string", "description": "City name"},
                    "units": {
                        "type": "string",
                        "enum": ["metric", "imperial"],
                        "description": "Units the temperature will be returned in",
                    },
                },
            },
        )

    async def execute(self, context: ToolContext, *, city: str, units: str):
        """
        This shows custom instrumentation that meshagent doesn't do automatically:
        1. Separate spans for validation, API call, parsing
        2. Custom attributes (API endpoint, response size)
        3. Events for important moments (rate limits etc.)
        4. Error handling with span status
        """
        log.info(f"Weather tool is running for city: {city} with units: {units}")
        calls.add(1, attributes={"city": city.lower(), "units": units})
        # Custom span for input validation
        with tracer.start_as_current_span("validate_input") as span:
            span.set_attribute("city", city)

            if not city or len(city) < 2:
                span.set_status(Status(StatusCode.ERROR, "Invalid city name"))
                span.add_event("validation_failed", {"reason": "city too short"})
                return {"error": "City name must be at least 2 characters"}

            span.add_event("validation_passed")

        # Custom span for external API call
        with tracer.start_as_current_span("fetch_weather_api") as span:
            # Add attributes about the API call
            api_url = f"https://wttr.in/{city}?format=j1"
            span.set_attribute("http.url", api_url)
            span.set_attribute("http.method", "GET")
            span.set_attribute("api.provider", "wttr.in")

            span.add_event("api_request_start")

            try:
                async with httpx.AsyncClient(timeout=10.0) as client:
                    response = await client.get(api_url)

                    # Record response attributes
                    span.set_attribute("http.status_code", response.status_code)
                    span.set_attribute("http.response_size", len(response.content))

                    if response.status_code == 429:
                        span.add_event("rate_limit_exceeded")
                        span.set_status(Status(StatusCode.ERROR, "Rate limited"))
                        return {"error": "Rate limit exceeded, try again later"}

                    response.raise_for_status()
                    data = response.json()

                    span.add_event(
                        "api_request_success",
                        {
                            "data_keys": list(data.keys()),
                        },
                    )

            except httpx.TimeoutException:
                span.set_status(Status(StatusCode.ERROR, "API timeout"))
                span.add_event("api_timeout")
                errors.add(1, attributes={"kind": "timeout", "city": city.lower()})
                return {"error": "Weather service timeout"}
            except Exception as e:
                errors.add(
                    1, attributes={"kind": type(e).__name__, "city": city.lower()}
                )
                span.set_status(Status(StatusCode.ERROR, str(e)))
                span.add_event("api_error", {"error_type": type(e).__name__})
                return {"error": f"Failed to fetch weather: {str(e)}"}

        # Custom span for parsing and formatting
        with tracer.start_as_current_span("parse_response") as span:
            try:
                current = data["current_condition"][0]
                location = data["nearest_area"][0]

                if units == "metric":
                    temperature = current["temp_C"]
                    degrees_in = "°C"
                elif units == "imperial":
                    temperature = current["temp_F"]
                    degrees_in = "°F"
                else:
                    log.warning(
                        f"Units {units} is not a valid unit. Must use metric or imperial"
                    )
                    return {"error": "Invalid units: must be 'metric' or 'imperial'"}

                result = {
                    "city": location["areaName"][0]["value"],
                    "country": location["country"][0]["value"],
                    "temperature": temperature,
                    "units": degrees_in,
                    "description": current["weatherDesc"][0]["value"],
                    "humidity": current["humidity"],
                    "wind_speed": current["windspeedKmph"],
                }

                # Record what we parsed
                span.set_attribute("parsed_fields", len(result))
                span.add_event("parse_success")

                return result

            except (KeyError, IndexError) as e:
                span.set_status(Status(StatusCode.ERROR, "Parse failed"))
                span.add_event("parse_error", {"error": str(e)})
                errors.add(1, attributes={"kind": "parse_error"})
                return {"error": "Failed to parse weather data"}


@service.path(path="/weather", identity="weather-toolkit")
class WeatherToolkit(RemoteToolkit):
    def __init__(self):
        super().__init__(
            name="weather-toolkit",
            title="Weather Toolkit",
            description="Tools for getting weather information",
            tools=[WeatherTool()],
        )


asyncio.run(service.run())

​
Custom spans

Use spans to track specific operations inside a trace. This example records:
  • execute.weather-toolkit.get_weather, which MeshAgent creates for the overall tool call
  • validate_input for city and units validation
  • fetch_weather_api for the outbound HTTP request, including attributes such as http.url, http.method, http.status_code, and http.response_size
  • parse_response for the final response shaping step
This gives you finer-grained visibility inside the spans MeshAgent already creates. Pattern:
Python
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("fetch_weather_api") as span:
    span.set_attribute("http.url", api_url)
    span.add_event("api_request_start")
    ...
    if resp.status_code == 429:
        span.set_status(Status(StatusCode.ERROR, "Rate limited"))
These spans appear in the room’s Traces tab.

​
Custom logs

otel_config() sets up an OTEL logging handler so normal logging calls are captured too.
Python
import logging
log = logging.getLogger(__name__)

log.info(f"Weather tool is running for city: {city} with units: {units}")
These logs appear in the room’s Logs tab.

​
Custom metrics

You can also add counters and histograms to track trends over time.
Python
from opentelemetry import metrics
meter = metrics.get_meter("weather.tools")

calls = meter.create_counter("weather.calls", unit="1",
                             description="Total weather tool invocations")

# When a call starts
calls.add(1, attributes={"city": city.lower(), "units": units})
These metrics appear in the room’s Metrics tab.

​
Deploy the sample

You can deploy this toolkit as a service and invoke get_weather() from Studio or the CLI.

​
Step 1: Package and deploy the room service

Package the sample with a meshagent.yaml file and a container image that MeshAgent can run. For the general deployment flow, see Deploy Services with MeshAgent. This example uses a MeshAgent runtime image plus a lightweight code image. The default YAML points at the public python-docs-examples image so you can run the docs example without building your own image first. Project structure: 
your-project/
├── Dockerfile                    # Shared by all samples
├── observability/
│   ├── observability.py
│   └── meshagent.yaml            # Config specific to this sample
└── another_sample/               # Other samples follow same pattern
    ├── another_sample.py
    └── meshagent.yaml
If you are building a single tool, you only need the observability/ folder.

​
Step 1a: Build a Docker image

Create a scratch Dockerfile and copy the files you want to run: 
FROM scratch

COPY . /
Build and push the image:
bash
docker buildx build . \
  -t "<REGISTRY>/<NAMESPACE>/<IMAGE_NAME>:<TAG>" \
  --platform linux/amd64 \
  --push

​
Step 1b: Define the service

Create a meshagent.yaml file that references:
  • Runtime image: The MeshAgent Python SDK image with all dependencies
  • Code mount: Your code-only image mounted at /src
  • Command path: Points to your sample’s specific location
kind: Service
version: v1
metadata:
  name: otel-example
  description: "An example weather tool with otel instrumentation"
ports:
- num: "*"
  liveness: "/"
  endpoints:
  - path: /weather
    meshagent:
      identity: weather-toolkit
container:
  image: "us-central1-docker.pkg.dev/meshagent-public/images/python-sdk:{SERVER_VERSION}-esgz"
  command: python /src/observability/observability.py
  storage:
    images:
      # Replace this image tag with your own code-only image if you build one.
      - image: "us-central1-docker.pkg.dev/meshagent-public/images/python-docs-examples:{SERVER_VERSION}"
        path: /src
        read_only: true
Path mapping:
  • Your code image contains /observability/observability.py
  • It’s mounted at /src in the runtime container
  • The command runs python /src/observability/observability.py
The default YAML in the docs uses us-central1-docker.pkg.dev/meshagent-public/images/python-docs-examples so you can test this example immediately without building your own image first. Replace this with your own image tag when deploying your code.

​
Step 1c: Deploy the service

From the directory that contains meshagent.yaml: 
meshagent service create --file "meshagent.yaml" --room=otel

​
Step 2: Invoke the tool

Once the service is deployed, invoke the tool from MeshAgent Studio or the CLI. You can also invoke the tool using the MeshAgent CLI:
bash
meshagent room agent invoke-tool \
  --room=otel \
  --toolkit=weather-toolkit \
  --tool=get_weather \
  --arguments='{"city":"Costa Mesa","units":"imperial"}'

​
View telemetry

In MeshAgent Studio, you can inspect the telemetry from this example in both the Session view and the Developer Console. After you invoke the weather tool, the Traces tab should show a tree like: 
execute.weather-toolkit.get_weather
 ├─ validate_input
 ├─ fetch_weather_api
 └─ parse_response
Logs and metrics appear under their respective tabs.

​
Next Steps

  • Observability: understand the built-in telemetry model and where to inspect it in MeshAgent Studio
  • Agents: understand how agents work in MeshAgent and start building your first one
  • Tools and Toolkits: learn how tools are discovered, shared, and called inside a MeshAgent room
  • Deploy Services with MeshAgent: deploy instrumented agents and tools as room or project services