Open-Dedalus API Documentation

Overview

Open-Dedalus is an open-source implementation of the Dedalus Labs SDK that provides a unified AI agents ecosystem with Bring Your Own Key (BYOK) support. It offers comprehensive integration with multiple AI providers, Model Context Protocol (MCP) servers, and advanced multi-model handoffs.

Table of Contents

• Core Classes
  • AsyncDedalus
  • DedalusRunner
  • RunResult
• Advanced Features
  • Multi-Model Handoffs
  • MCP Integration
  • Policy System
• Gateway Infrastructure
• Examples
• Provider Support

Core Classes

AsyncDedalus

The main client class for interacting with AI providers using your own API keys.

Constructor

AsyncDedalus(api_key: Optional[str] = None, base_url: Optional[str] = None)

Parameters:

• api_key (optional): API key for the provider. If not provided, will be read from environment variables.
• base_url (optional): Custom base URL for API calls.

Supported Providers:

• OpenAI (OPENAI_API_KEY)
• Anthropic (ANTHROPIC_API_KEY)
• Google Gemini (GOOGLE_API_KEY)
• Fireworks AI (FIREWORKS_API_KEY)
• xAI (XAI_API_KEY)
• Perplexity (PERPLEXITY_API_KEY)
• DeepSeek (DEEPSEEK_API_KEY)
• Groq (GROQ_API_KEY)
• Cohere (COHERE_API_KEY)
• Together AI (TOGETHERAPI_KEY)
• Cerebras (CEREBRAS_API_KEY)
• Mistral (MISTRAL_API_KEY)

Methods

chat_completion()

async chat_completion(
    model: str, 
    messages: List[dict], 
    **kwargs
) -> dict

Create a chat completion using the specified model.

Parameters:

• model: Model identifier (e.g., "openai/gpt-4o-mini", "claude-3-5-sonnet-20241022")
• messages: List of message dictionaries with "role" and "content" keys
• **kwargs: Additional parameters like temperature, max_tokens, etc.

Returns:

• Chat completion response dictionary in OpenAI-compatible format

Example:

import asyncio
from dedalus_labs import AsyncDedalus

async def main():
    client = AsyncDedalus()
    
    response = await client.chat_completion(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": "Hello, world!"}],
        temperature=0.7,
        max_tokens=1024
    )
    
    print(response["choices"][0]["message"]["content"])

asyncio.run(main())

close()

async close()

Close the HTTP session and cleanup resources.

Context Manager Support

AsyncDedalus supports async context managers for automatic resource cleanup:

async def main():
    async with AsyncDedalus() as client:
        response = await client.chat_completion(
            model="openai/gpt-4o-mini",
            messages=[{"role": "user", "content": "Hello!"}]
        )
        print(response["choices"][0]["message"]["content"])

DedalusRunner

The runner class for executing AI workflows with tool support, MCP integration, and multi-model handoffs.

Constructor

DedalusRunner(client: AsyncDedalus)

Parameters:

• client: AsyncDedalus client instance

Methods

run()

async run(
    input: str,
    model: Union[str, List[str]],
    tools: Optional[List[Callable]] = None,
    mcp_servers: Optional[List[str]] = None,
    stream: bool = False,
    policy: Optional[Callable] = None,
    on_tool_event: Optional[Callable] = None,
    max_steps: int = 10,
    handoff_strategy: Optional[HandoffStrategy] = None,
    **kwargs
) -> Union[RunResult, AsyncIterator[str]]

Run an AI workflow with the given input and configuration.

Parameters:

• input: User input/prompt
• model: Model(s) to use - string or list for handoffs
• tools: List of Python functions to use as tools
• mcp_servers: List of MCP server names to connect to
• stream: Whether to stream the response
• policy: Policy function for dynamic behavior control
• on_tool_event: Callback for tool execution events
• max_steps: Maximum number of execution steps (default: 10)
• handoff_strategy: Strategy for multi-model handoffs
• **kwargs: Additional parameters passed to the model

Returns:

• RunResult object or async iterator for streaming

Basic Example:

import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await runner.run(
        input="What is 2 + 2?",
        model="openai/gpt-4o-mini"
    )

    print(result.final_output)

asyncio.run(main())

Tools Example:

def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await runner.run(
        input="Calculate (15 + 27) * 2", 
        model="openai/gpt-4.1", 
        tools=[add, multiply]
    )

    print(result.final_output)

Streaming Example:

from dedalus_labs.utils.streaming import stream_async

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = runner.run(
        input="Tell me about AI",
        model="openai/gpt-4o-mini",
        stream=True
    )

    await stream_async(result)

RunResult

Data class containing the results of a runner execution.

Attributes

• final_output (str): The final response from the AI model
• steps (List[Dict[str, Any]]): List of execution steps with details
• model_used (str): The model that was used for the final response
• total_tokens (Optional[int]): Total tokens consumed (if available)

Advanced Features

Multi-Model Handoffs

Open-Dedalus supports intelligent routing between multiple models based on task complexity and model capabilities.

HandoffStrategy

from dedalus_labs import HandoffStrategy

# Available strategies:
HandoffStrategy.INTELLIGENT    # Automatic task analysis and model selection
HandoffStrategy.EXPERTISE      # Route based on model strengths
HandoffStrategy.SEQUENTIAL     # Step-by-step model progression
HandoffStrategy.COST_OPTIMIZED # Cost-performance optimization (framework)

Usage

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await runner.run(
        input="Find the year GPT-5 was released, then write a haiku about it.",
        model=["openai/gpt-4.1", "claude-3-5-sonnet-20241022"],
        mcp_servers=["tsion/brave-search-mcp"],
        handoff_strategy=HandoffStrategy.INTELLIGENT
    )

    print(result.final_output)

Task Types

The intelligent handoff system can detect these task types:

• SEARCH: Web search and information retrieval
• CREATIVE: Creative writing, poetry, storytelling
• CODE: Code generation and programming tasks
• MATH: Mathematical calculations and reasoning
• ANALYSIS: Data analysis and interpretation
• CONVERSATION: General conversation and Q&A

MCP Integration

Open-Dedalus provides comprehensive support for Model Context Protocol (MCP) servers.

Built-in MCP Servers

The framework includes configurations for these MCP servers:

• tsion/brave-search-mcp: Web search capabilities
• filesystem-mcp: File system operations
• echo-mcp: Echo and testing tools
• database-mcp: Database operations
• api-mcp: API integration tools
• development-mcp: Development utilities

Usage

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await runner.run(
        input="Who won Wimbledon 2025?",
        model="openai/gpt-4o-mini",
        mcp_servers=["tsion/brave-search-mcp"]
    )

    print(result.final_output)

MCP Client

from dedalus_labs.mcp import MCPClient

client = MCPClient()

# Connect to servers
await client.connect_servers(["tsion/brave-search-mcp"])

# Get available tools
tools = client.get_tool_schemas()

# Execute a tool
result = await client.execute_tool("search", {"query": "latest AI news"})

Policy System

The policy system allows dynamic behavior control at each execution step.

Policy Function

def policy(ctx: dict) -> dict:
    step = ctx.get("step", 1)
    
    # Apply different behavior based on step
    if step == 3:
        return {
            "message_prepend": [
                {"role": "system", "content": "You must speak like a pirate."}
            ]
        }
    
    # Set maximum steps
    return {"max_steps": 5}

Usage

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    def my_policy(ctx: dict) -> dict:
        step = ctx.get("step", 1)
        if step >= 3:
            return {"message_prepend": [{"role": "system", "content": "Be concise."}]}
        return {"max_steps": 4}

    result = await runner.run(
        input="Explain quantum computing",
        model="openai/gpt-4o-mini",
        policy=my_policy
    )

    print(result.final_output)

Gateway Infrastructure

Open-Dedalus includes a complete MCP gateway infrastructure for managing servers in production.

Components

MCPServerManager

Manages the lifecycle of MCP servers:

from dedalus_labs.gateway import MCPServerManager

manager = MCPServerManager()

# Start a server
await manager.start_server("brave-search", config={
    "command": "node",
    "args": ["dist/index.js"],
    "env": {"BRAVE_API_KEY": "your-key"}
})

# Monitor server health
health = await manager.get_server_health("brave-search")

# Stop a server
await manager.stop_server("brave-search")

MCPMarketplace

Discover and manage available MCP servers:

from dedalus_labs.gateway import MCPMarketplace

marketplace = MCPMarketplace()

# Search for servers
results = marketplace.search_servers(category="search", verified_only=True)

# Get popular servers
popular = marketplace.get_popular_servers()

# Get server details
server_info = marketplace.get_server_info("tsion/brave-search-mcp")

HealthMonitor

Monitor server health and performance:

from dedalus_labs.gateway import HealthMonitor

monitor = HealthMonitor()

# Start monitoring
monitor.start_monitoring("brave-search")

# Get health status
status = monitor.get_health_status("brave-search")

# Set up alerts
monitor.add_alert("brave-search", "memory", threshold=80)

Gateway API Server

The gateway provides a REST API for external integration:

from dedalus_labs.gateway.api import GatewayServer

server = GatewayServer()
await server.start(host="0.0.0.0", port=8080)

API Endpoints:

• GET /health - System health check
• GET /servers - List managed servers
• POST /servers - Start a new server
• POST /servers/{name}/start - Start server from marketplace
• POST /servers/{name}/stop - Stop server
• POST /servers/{name}/restart - Restart server
• GET /marketplace - Search marketplace
• GET /marketplace/categories - List categories
• GET /stats - Gateway statistics

Examples

Hello World

import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    response = await runner.run(
        input="What was the score of the 2025 Wimbledon final?",
        model="openai/gpt-4o-mini"
    )

    print(response.final_output)

asyncio.run(main())

Tool Chaining

import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner

def add(a: int, b: int) -> int:
    return a + b

def mul(x: int, y: int) -> int:
    return x * y

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await runner.run(
        input="1. Add 2 and 3. 2. Multiply that by 4. 3. Tell me what that number represents.",
        model="openai/gpt-4.1",
        tools=[add, mul],
        mcp_servers=["tsion/brave-search-mcp"]
    )

    print(result.final_output)

asyncio.run(main())

Complete Workflow

import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner, HandoffStrategy
from dedalus_labs.utils.streaming import stream_async

def calculate_tip(bill_amount: float, tip_percentage: float) -> float:
    """Calculate tip amount based on bill and percentage."""
    return bill_amount * (tip_percentage / 100)

async def main():
    async with AsyncDedalus() as client:
        runner = DedalusRunner(client)

        # Tool event handler
        def on_tool_event(event):
            print(f"[TOOL] {event['type']}: {event.get('function', '')} -> {event.get('result', event.get('error', ''))}")

        # Policy for dynamic behavior
        def restaurant_policy(ctx):
            step = ctx.get("step", 1)
            if step >= 3:
                return {"message_prepend": [{"role": "system", "content": "Provide detailed explanations."}]}
            return {"max_steps": 5}

        result = await runner.run(
            input="Calculate a 18% tip on a \$67.50 restaurant bill, then search for good restaurants nearby and summarize the top recommendation",
            model=["openai/gpt-4.1", "claude-3-5-sonnet-20241022"],
            tools=[calculate_tip],
            mcp_servers=["tsion/brave-search-mcp"],
            handoff_strategy=HandoffStrategy.INTELLIGENT,
            policy=restaurant_policy,
            on_tool_event=on_tool_event,
            temperature=0.7
        )

        print(f"\nFinal Result:\n{result.final_output}")
        print(f"\nUsed Model: {result.model_used}")
        print(f"Total Steps: {len(result.steps)}")

asyncio.run(main())

Provider Support

Model Naming Convention

Use the format provider/model for explicit provider selection:

• OpenAI: openai/gpt-4o-mini, openai/gpt-4.1
• Anthropic: anthropic/claude-3-5-sonnet-20241022
• Google: google/gemini-pro
• Others: fireworks/llama-v3-70b-instruct, groq/mixtral-8x7b-32768

Auto-Detection

Models without provider prefix are auto-detected:

• gpt-4o-mini → OpenAI
• claude-3-5-sonnet-20241022 → Anthropic
• gemini-pro → Google

Environment Variables

Set the appropriate environment variable for your provider:

export OPENAI_API_KEY="your-openai-key"
export ANTHROPIC_API_KEY="your-anthropic-key" 
export GOOGLE_API_KEY="your-google-key"
# ... other providers

Error Handling

API Errors

try:
    response = await client.chat_completion(
        model="invalid/model",
        messages=[{"role": "user", "content": "Hello"}]
    )
except Exception as e:
    print(f"API error: {e}")

Tool Execution Errors

Tool errors are automatically handled and reported to the model:

def divide(a: float, b: float) -> float:
    """Divide two numbers."""
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

# The runner will catch the error and continue execution
result = await runner.run(
    input="Divide 10 by 0",
    model="openai/gpt-4o-mini",
    tools=[divide]
)

MCP Connection Errors

MCP connection failures are logged but don't stop execution:

result = await runner.run(
    input="Search for something",
    model="openai/gpt-4o-mini",
    mcp_servers=["invalid-server"]  # Will log error but continue
)

Best Practices

Resource Management

Always use context managers or explicit cleanup:

# Preferred: Context manager
async with AsyncDedalus() as client:
    # Your code here
    pass

# Alternative: Explicit cleanup
client = AsyncDedalus()
try:
    # Your code here
    pass
finally:
    await client.close()

Tool Design

Design tools with clear docstrings and type hints:

def weather_info(location: str, units: str = "celsius") -> str:
    """
    Get weather information for a specific location.
    
    Args:
        location: City name or coordinates
        units: Temperature units (celsius/fahrenheit)
    
    Returns:
        Weather information string
    """
    # Implementation here
    pass

Error Recovery

Implement graceful error handling:

async def safe_runner_execution():
    try:
        result = await runner.run(
            input="Complex task",
            model="openai/gpt-4o-mini",
            max_steps=5
        )
        return result.final_output
    except Exception as e:
        logger.error(f"Runner execution failed: {e}")
        return "Sorry, I couldn't complete that task."

Dependencies

Required

• Python 3.8+
• asyncio (standard library)
• json (standard library)
• inspect (standard library)

Optional

• aiohttp: For HTTP functionality (recommended)
• python-dotenv: For environment variable management

Installation

# Basic installation
pip install open-dedalus

# With optional dependencies
pip install open-dedalus[full]

# Development installation
pip install -e .[dev]

This comprehensive API documentation covers all major components and features of the Open-Dedalus SDK. For additional examples and advanced usage patterns, refer to the examples directory in the repository.