Evidor

Evidor is an open-source, provider-agnostic runtime for building AI agents. It provides a clean, unified interface for conversational agents across multiple model providers, with built-in context window management and automatic summarization.

Install

Install the provider adapter(s) you need:

pip install "evidor[openai]"
# or: evidor[anthropic], evidor[gemini], evidor[mcp], evidor[all]

Quickstart

from evidor import Agent, OpenAIProvider

agent = Agent(OpenAIProvider(model="gpt-4.1-mini"))
response = agent.send("Give a one-sentence explanation of dependency inversion.")
print(response.text)

Core Usage

Multi-turn Conversations

The Agent maintains session state across conversation turns. Use send() to advance the conversation:

from evidor import Agent, AnthropicProvider

agent = Agent(AnthropicProvider(model="claude-3-5-sonnet-20241022"))

agent.send("My favorite color is navy blue.")
response = agent.send("What color did I say I like?")
print(response.text)

System Prompts

You can configure an initial system prompt when creating an Agent. The system prompt is preserved across conversation compaction and history resets:

from evidor import Agent, GeminiProvider

agent = Agent(
    GeminiProvider(model="gemini-2.5-flash"),
    system_prompt="You are a concise technical writer. Avoid buzzwords and respond in bullet points.",
)

response = agent.send("How does a TCP handshake work?")
print(response.text)

Context Window Management & Compaction

Agent automatically manages message history to stay within token and message limits using ConversationContext:

from evidor import DEFAULT_CONTEXT_WINDOW, DEFAULT_MAX_MESSAGES, Agent, OpenAIProvider

agent = Agent(
    OpenAIProvider(model="gpt-4.1-mini"),
    context_window=16_000,  # Max token budget (default: 16,000)
    max_messages=50,        # Max messages retained before compaction (default: 50)
)

When conversation history exceeds max_messages or the estimated token budget:

• Older messages are automatically summarized using the configured model provider.
• The summary is injected as a bounded summary message (Message(role="system", ..., is_summary=True)).
• The permanent system prompt and recent conversational turns remain intact.

Configurable Summarization Model

By default, the agent uses its primary provider and model for summarization during compaction. You can configure a specific model name string or a separate ModelProvider instance (e.g. to use a faster, lighter model for background summaries):

from evidor import Agent, GeminiProvider, OpenAIProvider

# Option 1: Configure a different model name on the same provider
agent = Agent(
    OpenAIProvider(model="gpt-4.1"),
    summarization_model="gpt-4.1-mini",
)

# Option 2: Provide an entirely separate ModelProvider instance
agent = Agent(
    OpenAIProvider(model="gpt-4.1"),
    summarization_model=GeminiProvider(model="gemini-2.5-flash"),
)

Inspecting and Clearing History

You can inspect the retained conversation history or reset turns at any time:

# Inspect current history (tuple of Message objects)
for message in agent.messages:
    prefix = "[SUMMARY] " if message.is_summary else ""
    print(f"{prefix}{message.role}: {message.content}")

# Clear conversation turns while keeping the original system prompt
agent.clear_history()

Low-level Generation with GenerationRequest

For direct provider calls without stateful session management, instantiate messages and requests directly:

from evidor import GenerationRequest, Message, OpenAIProvider

provider = OpenAIProvider(model="gpt-4.1-mini")

# Multi-message request
request = GenerationRequest(
    messages=[
        Message(role="system", content="You are a helpful assistant."),
        Message(role="user", content="Explain Raft consensus in one paragraph."),
    ]
)
response = provider.generate(request)
print(response.text)

# Single-prompt compatibility shortcut
simple_request = GenerationRequest(prompt="Hello!")
response = provider.generate(simple_request)

Tools & Function Calling

Evidor provides first-class primitives for defining tools and equipping agents with automatic function-calling execution loops.

Defining Tools with @tool

Transform any Python function into a provider-neutral tool using the @tool decorator. Evidor automatically derives the JSON Schema for parameters from type annotations and extracts descriptions from docstrings (Google, Sphinx, or plain styles):

from evidor import tool

@tool
def get_weather(location: str, unit: str = "celsius") -> str:
    """Get the current weather forecast for a given location.

    Args:
        location: City and country or state, e.g. 'San Francisco, CA'.
        unit: Temperature scale ('celsius' or 'fahrenheit').
    """
    return f"Weather in {location}: 22° {unit}, clear skies."

@tool(name="calc_add", description="Add two numbers together.")
def add(a: float, b: float) -> float:
    return a + b

Equipping Agents with Tools

Pass your tools directly into Agent(..., tools=[...]). When a model decides to call one or more tools, the agent automatically executes them, feeds the results back to the model as role="tool" messages, and loops until the model generates a final response:

from evidor import Agent, GeminiProvider, tool

@tool
def search_database(query: str) -> list[str]:
    """Search internal records by keyword."""
    return [f"Record 1 matching '{query}'", f"Record 2 matching '{query}'"]

agent = Agent(
    GeminiProvider(model="gemini-2.5-flash"),
    tools=[search_database],
    max_tool_iterations=10,  # Maximum tool execution turns per send() (default: 10)
)

response = agent.send("Can you check our records for 'alpha project'?")
print(response.text)

Async Tools & Execution Timeouts

Evidor seamlessly supports asynchronous functions and execution timeouts:

import asyncio
from evidor import Agent, GeminiProvider, tool

# Async functions are automatically executed without blocking or crashing
@tool(timeout=5.0)  # Timeout in seconds
async def fetch_webpage(url: str) -> str:
    """Fetch content from a URL."""
    await asyncio.sleep(0.1)
    return f"Contents of {url}"

# Configure an agent with a default timeout for all its tools
agent = Agent(
    GeminiProvider(model="gemini-2.5-flash"),
    tools=[fetch_webpage],
    tool_timeout=10.0,      # Default tool timeout in seconds
    max_tool_iterations=10, # Max tool turns before forcing final synthesis
)

If a tool times out, raises an error, or if the model emits malformed JSON arguments, Evidor safely captures the error and feeds it back to the model as a role="tool" message so the LLM can self-correct.

Native Asynchronous Conversations (send_async)

In async environments (like FastAPI, Tornado, or asynchronous data pipelines), use await agent.send_async(...). When called, all async tools execute natively on the caller's event loop without thread transitions, cleanly preserving shared connection pools, locks, and contextvars:

import asyncio
from evidor import Agent, OpenAIProvider, tool

@tool
async def lookup_account(account_id: str) -> dict:
    return {"id": account_id, "status": "active"}

async def main() -> None:
    agent = Agent(OpenAIProvider(model="gpt-4.1-mini"), tools=[lookup_account])
    response = await agent.send_async("Check status for account 1234")
    print(response.text)

asyncio.run(main())

Manual Tool Definition

For dynamic or programmatic tools without a standard Python function signature, instantiate Tool directly:

from evidor import Tool

custom_tool = Tool(
    name="query_sql",
    description="Run a read-only SQL query.",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "SQL statement"}
        },
        "required": ["query"],
    },
    func=lambda query: f"Results for: {query}",
    timeout=5.0,
)

Evidor's tool calling works consistently across OpenAI, Anthropic, and Gemini, adapting each provider's underlying tool schema and message format transparently.

Built-in Tools

Evidor includes a couple of dependency-free tools you can opt into. They are not attached to agents automatically:

from evidor import Agent, OpenAIProvider, calculator, get_current_time

agent = Agent(
    OpenAIProvider(model="gpt-4.1-mini"),
    tools=[calculator, get_current_time],
)

get_current_time returns an ISO 8601 timestamp and accepts an IANA timezone name (UTC by default). calculator evaluates basic arithmetic without executing Python code.

Create filesystem tools scoped to an existing directory for listing, searching, reading, creating, updating, and deleting files. Paths supplied by the model are resolved under that root, and file reads and writes and result counts have configurable limits:

from evidor import Agent, OpenAIProvider, calculator, filesystem_tools, get_current_time

agent = Agent(
    OpenAIProvider(model="gpt-4.1-mini"),
    tools=[
        calculator,
        get_current_time,
        *filesystem_tools("./project", max_file_bytes=100_000, max_results=100),
    ],
)

filesystem_tools returns six tools:

• list_files lists files in the root or a subdirectory.
• read_file reads a UTF-8 text file.
• search_files searches UTF-8 text files for a literal string.
• create_file creates a new UTF-8 text file; it fails if the path already exists. Its parent directory must already exist.
• write_file replaces the contents of an existing UTF-8 text file.
• delete_file deletes an existing file (directories cannot be deleted).

All paths stay within the configured root. max_file_bytes limits the size of a file read or written (including initial content passed to create_file); max_results bounds listing and search results, and max_files_scanned bounds recursive search work.

Web Search

Use web_search() to create a provider-neutral agent tool. The tool depends only on the WebSearchProvider interface, so the selected search service can be changed without changing the agent setup. The bundled Tavily, Exa, and Brave adapters use only Python's standard library; set the provider's API key environment variable (or pass api_key= explicitly).

export TAVILY_API_KEY="..."
# or EXA_API_KEY="..." / BRAVE_SEARCH_API_KEY="..."

from evidor import Agent, OpenAIProvider, TavilySearchProvider, web_search

search = web_search(TavilySearchProvider(), max_results=5)
agent = Agent(OpenAIProvider(model="gpt-4.1-mini"), tools=[search])

response = agent.send("Find the current Python release notes and cite the sources.")
print(response.text)

Switch providers by replacing TavilySearchProvider() with ExaSearchProvider() or BraveSearchProvider(). To use another service, implement the small WebSearchProvider contract: search(query: str, *, max_results: int) -> list[SearchResult]; both types are exported from evidor.

Model Context Protocol (MCP)

Evidor includes first-class support for the Model Context Protocol (MCP). Agents can connect to any MCP server (over local stdio, modern Streamable HTTP, or legacy SSE), automatically discover tools, execute them, and feed responses back into the model.

Install MCP support:

pip install "evidor[mcp]"

Quickstart: Connecting an Agent to an MCP Server

You can pass mcp_servers directly when instantiating an Agent. For example, integrating an Amazon MCP server (e.g. AWS Bedrock, Amazon search, or a custom store MCP server):

from evidor import Agent, OpenAIProvider

agent = Agent(
    OpenAIProvider(model="gpt-4.1-mini"),
    mcp_servers={
        "amazon": {
            "command": "uvx",
            "args": ["amazon-mcp-server"],
            "env": {"AWS_PROFILE": "default"},
        }
    },
)

# Tools exposed by the Amazon MCP server are automatically discovered and made available
response = agent.send("Search for Kindle Paperwhite on Amazon and give me a summary.")
print(response.text)

# Cleanly close subprocesses when done
agent.close()

Or using a context manager:

with Agent(provider, mcp_servers={"amazon": {"command": "uvx", "args": ["amazon-mcp-server"]}}) as agent:
    response = agent.send("Find top noise-cancelling headphones on Amazon.")
    print(response.text)

Async MCP with send_async

In asynchronous applications, use send_async() and async with:

import asyncio
from evidor import Agent, AnthropicProvider

async def main():
    async with Agent(
        AnthropicProvider(model="claude-3-5-sonnet-20241022"),
        mcp_servers={
            "amazon": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-amazon"],
            }
        },
    ) as agent:
        response = await agent.send_async("Check the status of order ORD-12345.")
        print(response.text)

asyncio.run(main())

Remote & Hosted MCP Servers (Streamable HTTP & SSE)

Evidor natively supports connecting to remote, hosted MCP servers. In modern MCP specifications, Streamable HTTP (streamable-http or http) is the primary standard for HTTP-based communication, while legacy SSE is supported for backwards compatibility:

from evidor import Agent, OpenAIProvider, MCPServerConfig

# Modern Streamable HTTP (recommended for hosted MCP servers)
agent = Agent(
    OpenAIProvider("gpt-4.1-mini"),
    mcp_servers={
        "hosted-aws": {
            "url": "https://mcp.my-hosted-aws.com/mcp",
            "transport": "streamable-http",  # default for non-/sse URLs
            "headers": {"Authorization": "Bearer YOUR_API_KEY"},
        }
    },
)

# Or explicitly using factory methods:
config = MCPServerConfig.http(
    name="hosted-service",
    url="https://mcp.service.internal/mcp",
    headers={"x-api-key": "..."},
)

Connecting Multiple MCP Servers

Connect multiple MCP servers simultaneously. Evidor automatically prevents tool naming collisions across servers:

from evidor import Agent, GeminiProvider, MCPServerConfig

agent = Agent(
    GeminiProvider(model="gemini-2.5-flash"),
    mcp_servers=[
        # Stdio server
        MCPServerConfig.stdio(
            name="amazon",
            command="uvx",
            args=["amazon-mcp-server"],
        ),
        # Remote Streamable HTTP server
        MCPServerConfig.http(
            name="remote-inventory",
            url="https://inventory.internal/mcp",
            headers={"Authorization": "Bearer secret-token"},
        ),
        # Legacy SSE server
        MCPServerConfig.sse(
            name="weather",
            url="http://localhost:8080/sse",
        ),
    ],
)

response = agent.send("What is the weather in Seattle and recommend waterproof hiking boots on Amazon?")
print(response.text)

mcp_servers vs mcp_client: When to Use Which

An Agent supports two parameters for MCP integration depending on whether the agent manages the connection lifecycle:

Parameter	What You Pass	Best For
mcp_servers	Raw configuration (dicts, commands, URLs, or config files)	Standard applications where the agent owns the server processes. The agent automatically boots the client, connects to servers, and terminates subprocesses when closed.
mcp_client	An already-instantiated MCPClient instance	Shared connections / multi-agent systems (e.g. FastAPI backends) where a single client is kept open and shared across multiple agents without spawning duplicate subprocesses.

Using mcp_client to Share Connections Across Agents

from evidor import Agent, MCPClient, OpenAIProvider

# Boot a single shared client for your application
shared_mcp = MCPClient.from_file("claude_desktop_config.json")
shared_mcp.connect()

# Both agents reuse the exact same client without duplicating subprocesses:
researcher = Agent(OpenAIProvider("gpt-4.1-mini"), mcp_client=shared_mcp)
assistant = Agent(OpenAIProvider("gpt-4.1-mini"), mcp_client=shared_mcp)

response1 = researcher.send("Search records for client ABC.")
response2 = assistant.send("Verify credentials for client ABC.")

# Clean up once when the application terminates
shared_mcp.close()

Advanced: Standalone MCPClient and Tool Composition

You can also use MCPClient standalone to inspect tools, retrieve resources, or mix MCP tools with standard Python tools:

import asyncio
from evidor import Agent, MCPClient, OpenAIProvider, tool

@tool
def internal_calculator(x: int, y: int) -> int:
    return x * y

async def main():
    async with MCPClient({"amazon": {"command": "uvx", "args": ["amazon-mcp-server"]}}) as mcp:
        # Retrieve all Evidor Tool objects from the MCP server
        mcp_tools = await mcp.get_tools_async()

        # Combine MCP tools with local tools
        agent = Agent(OpenAIProvider("gpt-4.1-mini"), tools=[*mcp_tools, internal_calculator])
        response = await agent.send_async("Calculate total price for 3 units of item B08N5WRWNW.")
        print(response.text)

asyncio.run(main())

Providers

Supported providers and their typical models:

Provider	Example Model	Environment Variable	Extra Install
OpenAIProvider	gpt-4.1-mini	OPENAI_API_KEY	evidor[openai]
AnthropicProvider	claude-3-5-sonnet-20241022	ANTHROPIC_API_KEY	evidor[anthropic]
GeminiProvider	gemini-2.5-flash	GEMINI_API_KEY	evidor[gemini]

API keys can be supplied explicitly via the api_key constructor argument or read automatically from environment variables.

Custom Providers

Agent depends only on the ModelProvider protocol. To support another LLM or backend, implement the generate(request) method:

from evidor import GenerationRequest, GenerationResponse, ModelProvider

class CustomProvider:
    def __init__(self, model: str = "custom-model") -> None:
        self.model = model

    def with_model(self, model: str) -> "CustomProvider":
        return CustomProvider(model=model)

    def generate(self, request: GenerationRequest) -> GenerationResponse:
        prompt_text = request.prompt
        # or iterate over request.messages
        return GenerationResponse(text="Custom response", model=self.model)

Build and release

Create distributable artifacts locally:

python -m pip install -e ".[dev]"
python -m pytest
python -m build
python -m twine check dist/*

Releases are performed by GitHub Actions and use Conventional Commits to select the next version:

• fix: ... creates a patch release.
• feat: ... creates a minor release.
• feat!: ... or a BREAKING CHANGE: footer creates a major release.

Pushes to dev publish -dev.N prereleases to TestPyPI. Pushes to main publish stable releases to PyPI. Configure TEST_PYPI_API_TOKEN as a repository or dev environment secret, and configure PyPI trusted publishing for the prod environment before enabling the workflow.