Model Context Protocol (MCP)

pip install netra-sdk mcp

from netra import Netra
from mcp.server import Server
from mcp.server.stdio import stdio_server
import os

# Initialize Netra
Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True
)

# Create MCP server - automatically traced
server = Server("example-server")

from netra.decorators import workflow, task
from netra import SpanWrapper
from mcp.server.models import Tool
from mcp.types import TextContent

@server.call_tool()
async def execute_tool(name: str, arguments: dict):
    span = SpanWrapper("mcp-tool", {
        "tool.name": name
    }).start()
    
    try:
        result = await process_tool(name, arguments)
        span.set_attribute("tool.result", str(result))
        span.end()
        
        return [TextContent(
            type="text",
            text=str(result)
        )]
    except Exception as e:
        span.set_attribute("error", str(e))
        span.end()
        raise

@task()
async def process_tool(name: str, args: dict):
    # Tool implementation
    return {"status": "success"}

from netra import SpanWrapper
from mcp.server.models import Resource
from mcp.types import TextResourceContents

@server.list_resources()
async def list_resources():
    return [
        Resource(
            uri="file:///example.txt",
            name="Example Resource"
        )
    ]

@server.read_resource()
async def read_resource(uri: str):
    span = SpanWrapper("mcp-resource", {
        "resource.uri": uri
    }).start()
    
    try:
        content = await load_resource(uri)
        span.set_attribute("resource.size", len(content))
        span.end()
        
        return TextResourceContents(
            uri=uri,
            mimeType="text/plain",
            text=content
        )
    except Exception as e:
        span.set_attribute("error", str(e))
        span.end()
        raise

from netra import SpanWrapper
from mcp.server.models import Prompt, PromptMessage

@server.list_prompts()
async def list_prompts():
    return [
        Prompt(
            name="example-prompt",
            description="Example prompt template"
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
    span = SpanWrapper("mcp-prompt", {
        "prompt.name": name
    }).start()
    
    messages = generate_prompt(name, arguments)
    span.set_attribute("prompt.messages", len(messages))
    span.end()
    
    return messages

from netra import Netra

Netra.init(
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}",
    trace_content=True,
    app_name="mcp-server"
)