Model Context Protocol (MCP)

pip install netra-sdk mcp

npm install netra-sdk @modelcontextprotocol/sdk

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

import { Netra } from "netra-sdk";
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

// Initialize Netra
await Netra.init({
  headers: `x-api-key=${process.env.NETRA_API_KEY}`,
  traceContent: true
});

// Create MCP server - automatically traced
const server = new Server(
  {
    name: "example-server",
    version: "1.0.0"
  },
  {
    capabilities: {
      tools: {},
      resources: {}
    }
  }
);

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"}

import { workflow, task, SpanWrapper } from "netra-sdk";
import { CallToolRequestSchema } from "@modelcontextprotocol/sdk/types.js";

// Register tool with tracing
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const span = new SpanWrapper("mcp-tool", {
    "tool.name": request.params.name
  }).start();
  
  try {
    const result = await executeTool(request.params);
    span.setAttribute("tool.result", JSON.stringify(result));
    span.end();
    
    return {
      content: [{ type: "text", text: JSON.stringify(result) }]
    };
  } catch (error) {
    span.setAttribute("error", String(error));
    span.end();
    throw error;
  }
});

@task()
async function executeTool(params: any) {
  // 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

import { ReadResourceRequestSchema } from "@modelcontextprotocol/sdk/types.js";

server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const span = new SpanWrapper("mcp-resource", {
    "resource.uri": request.params.uri
  }).start();
  
  try {
    const content = await loadResource(request.params.uri);
    span.setAttribute("resource.size", content.length);
    span.end();
    
    return {
      contents: [{
        uri: request.params.uri,
        mimeType: "text/plain",
        text: content
      }]
    };
  } catch (error) {
    span.setAttribute("error", String(error));
    span.end();
    throw error;
  }
});

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

import { GetPromptRequestSchema } from "@modelcontextprotocol/sdk/types.js";

server.setRequestHandler(GetPromptRequestSchema, async (request) => {
  const span = new SpanWrapper("mcp-prompt", {
    "prompt.name": request.params.name
  }).start();
  
  const prompt = generatePrompt(request.params);
  span.setAttribute("prompt.messages", prompt.messages.length);
  span.end();
  
  return prompt;
});

from netra import Netra

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

import { Netra, NetraInstruments } from "netra-sdk";

await Netra.init({
  headers: `x-api-key=${process.env.NETRA_API_KEY}`,
  traceContent: true,
  appName: "mcp-server"
});