Python SDK

pip install hmdl

from hmdl import HeimdallClient, trace_mcp_tool

# Initialize with explicit configuration
client = HeimdallClient(
    endpoint="http://localhost:4318",
    org_id="your-org-id",
    project_id="your-project-id",
    service_name="my-mcp-server",
    environment="development"
)

@trace_mcp_tool()
def search_documents(query: str, limit: int = 10) -> dict:
    """Search for documents matching the query."""
    results = perform_search(query, limit)
    return {
        "results": results,
        "query": query,
        "total": len(results)
    }

# Call your tool normally
result = search_documents("machine learning", limit=5)

@trace_mcp_tool("document-search")
def search(query: str) -> dict:
    return {"results": []}

# At the end of your application
client.flush()

from hmdl import observe

@observe(name="fetch-user-data")
def fetch_user(user_id: str) -> dict:
    # Fetch user from database
    return {"id": user_id, "name": "John"}

@observe(capture_output=False)  # Don't capture sensitive output
def process_payment(card_number: str, amount: float) -> bool:
    # Process payment
    return True

@trace_mcp_tool()
async def async_search(query: str) -> dict:
    results = await async_perform_search(query)
    return {"results": results}

# Async flush
await client.flush_async()

@trace_mcp_tool()
def risky_operation(data: str) -> dict:
    if not data:
        raise ValueError("Data cannot be empty")
    return {"processed": data}

try:
    result = risky_operation("")
except ValueError:
    pass  # Error is captured in the trace

client.flush()  # Don't forget to flush!

import os
from hmdl import HeimdallClient, trace_mcp_tool, observe

# Configure via environment variables
os.environ["HEIMDALL_ENDPOINT"] = "http://localhost:4318"
os.environ["HEIMDALL_ORG_ID"] = "my-org"
os.environ["HEIMDALL_PROJECT_ID"] = "my-project"

# Initialize client
client = HeimdallClient(
    service_name="document-mcp-server",
    environment="production"
)

# Your MCP tools
@trace_mcp_tool()
def search_documents(query: str, limit: int = 10) -> dict:
    """Search for documents."""
    results = db.search(query, limit)
    return {"results": results, "count": len(results)}

@trace_mcp_tool()
def get_document(doc_id: str) -> dict:
    """Retrieve a specific document."""
    doc = db.get(doc_id)
    if not doc:
        raise ValueError(f"Document {doc_id} not found")
    return doc

@trace_mcp_tool()
def create_document(title: str, content: str) -> dict:
    """Create a new document."""
    doc = db.create(title=title, content=content)
    return {"id": doc.id, "created": True}

# Internal helper with tracing
@observe(name="validate-query")
def validate_query(query: str) -> bool:
    return len(query) >= 3

# Main execution
if __name__ == "__main__":
    try:
        # Run your MCP server logic
        result = search_documents("test query", limit=5)
        print(result)
    finally:
        # Always flush before exit
        client.flush()