Async Support

import mantisdk.tracing as tracing
 
@tracing.trace
async def async_process(query: str) -> dict:
    """Async function - decorator works automatically."""
    result = await some_async_operation(query)
    return {"result": result}
 
# Usage
result = await async_process("hello")

import mantisdk.tracing as tracing
 
async def my_workflow(query: str):
    async with tracing.atrace("async-workflow", input=query) as span:
        result = await async_operation()
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
        return result

import mantisdk.tracing as tracing
 
async def my_workflow(query: str):
    async with tracing.atrace("workflow", input=query):
        # Async child span
        async with tracing.aspan("fetch_data") as s:
            data = await fetch_from_api()
            s.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, str(len(data)))
 
        # Another async child span
        async with tracing.aspan("process_data") as s:
            result = await process(data)
            s.set_attribute(tracing.semconv.OBSERVATION_OUTPUT, result)
 
        return result

import mantisdk.tracing as tracing
 
async def hybrid_workflow(query: str):
    # Async trace
    async with tracing.atrace("workflow", input=query) as root:
        # Sync span (still works in async context)
        with tracing.span("sync_operation") as s:
            sync_result = sync_function()
            s.set_attribute("sync_result", sync_result)
 
        # Async span
        async with tracing.aspan("async_operation") as s:
            async_result = await async_function()
            s.set_attribute("async_result", async_result)
 
        root.set_attribute(tracing.semconv.TRACE_OUTPUT, async_result)
        return async_result

async def main():
    # atrace for the root
    async with tracing.atrace("main_workflow") as root:
        # aspan for children
        async with tracing.aspan("step_1"):
            await step_one()
 
        async with tracing.aspan("step_2"):
            await step_two()

async with tracing.atrace(
    name,                        # Span name (required)
    input=None,                  # Input value
    kind=SpanKind.INTERNAL,      # OpenTelemetry span kind
    attributes=None,             # Initial attributes dict
    record_exception=True,       # Record exceptions on span
    set_status_on_exception=True # Set ERROR status on exception
) as span:
    ...

import asyncio
import mantisdk.tracing as tracing
 
async def fetch_all(urls: list[str]):
    async with tracing.atrace("fetch_all", input=str(urls)):
        # Run concurrent fetches
        tasks = []
        for url in urls:
            tasks.append(fetch_url(url))
 
        results = await asyncio.gather(*tasks)
        return results
 
async def fetch_url(url: str):
    """Each concurrent task gets its own span."""
    async with tracing.aspan(f"fetch:{url}", input=url) as span:
        result = await http_get(url)
        span.set_attribute("status_code", result.status)
        return result

#!/usr/bin/env python3
"""Complete async tracing example."""
 
import asyncio
from dotenv import load_dotenv
load_dotenv()
 
import mantisdk.tracing as tracing
 
 
@tracing.trace
async def classify_query(query: str) -> str:
    """Async classification - decorator auto-captures I/O."""
    await asyncio.sleep(0.1)  # Simulate async work
    if "weather" in query.lower():
        return "weather"
    return "general"
 
 
@tracing.trace
async def generate_response(query: str, query_type: str) -> str:
    """Async generation - decorator auto-captures I/O."""
    await asyncio.sleep(0.2)  # Simulate LLM call
    return f"Response for {query_type} query: {query}"
 
 
async def agent_workflow(user_id: str, query: str):
    """Main async workflow with explicit spans."""
    async with tracing.atrace("agent_session", input=query) as root:
        # Set user context
        root.set_attribute(tracing.semconv.USER_ID, user_id)
 
        # Classify (uses decorator, auto-traced)
        query_type = await classify_query(query)
 
        # Generate (uses decorator, auto-traced)
        response = await generate_response(query, query_type)
 
        # Set output
        root.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
 
        return response
 
 
async def main():
    tracing.init(service_name="async-agent")
 
    try:
        result = await agent_workflow(
            user_id="user-123",
            query="What's the weather like?"
        )
        print(f"Result: {result}")
    finally:
        tracing.shutdown()
 
 
if __name__ == "__main__":
    asyncio.run(main())