PyStreamMCP

Context and token optimization for AI agents — query planning, source discovery, and MCP orchestration.

What it is

PyStreamMCP is an AI-agent intelligence layer: query planning, context discovery, and token-cost optimization, exposed both as a Python SDK and as an MCP (Model Context Protocol) tool server. It also includes a small event-driven orchestration/federation layer for coordinating multiple MCP endpoints. The published package is pure Python — pip install works with no Rust toolchain or compiler.

Features / Highlights

• Agent query planning — pystreammcp.Agent with selectable optimization strategies (token_efficient, balanced, quality_first) and token accounting.
• Source discovery — SourceRegistry ranks registered data sources by real word-overlap relevance (Jaccard) to a query; backs REST /discover, the pystreammcp_discover MCP tool, and LangchainAdapter.discover().
• MCP tool server — PyStreamMCPHandler wires the MCP tool-call surface directly to a live Orchestrator/EventRouter; MCP tool args are validated with Pydantic and the advertised schema is generated from the same models.
• MCP orchestration & federation — Orchestrator probes each configured endpoint's real tools/list MCP method, with lexical tool-relevance ranking and capability matching.
• Webhook event server — pystreammcp.server.create_flask_app() exposes an HMAC-SHA256-signed event endpoint; unsigned or invalid events are rejected, and the endpoint fails closed (503) if no secret is configured.
• Framework adapters — real-tested LangChain, LlamaIndex, and Semantic Kernel adapters under pystreammcp.integrations.
• CLI — pystreammcp with query / server / version / dashboard.
• OKF integration — an Open Knowledge Format catalog/discovery/query-planner feature (see OKF_INTEGRATION.md).

Who it's for / Use cases

• Giving an AI agent a query-planning layer over multiple data sources, exposed as real MCP tools rather than hand-rolled per-agent glue code.
• Coordinating multiple MCP endpoints through the event-driven orchestration/federation layer instead of wiring each one separately.
• Tracking token-cost optimization across agent queries.
• Adding MCP support to an existing agent framework through the shipped adapters.

What it is not

• The Rust backend is not shipped. core/ and python/src/lib.rs are an in-progress Rust workspace intended as a future performance backend; the wheel built from pyproject.toml is pure Python. The workspace compiles and its tests pass, but it is tracked separately and not part of the published package.
• Some orchestration capabilities are not fully implemented. Cross-project federated query execution and cross-database joins currently report "status": "not_implemented" with an explanation, rather than fabricated result rows.
• Source discovery is a word-overlap heuristic, not semantic search. SourceRegistry uses Jaccard token overlap; wrap it with your own similarity scoring if you need embeddings.
• Some framework names in the code are experimental helpers, not integrations. crewai, pydantic_ai, haystack, temporal, and airflow modules exist under pystreammcp.integrations/pystreammcp.orchestration, but they never import those vendor SDKs and are not supported. See Experimental / unsupported integrations.
• Optional concurrency/framework packages have Python constraints. The core SDK (Agent, SourceRegistry, Orchestrator) supports Python 3.9+, but the mcp and semantic-kernel extras require Python 3.10+ (their upstream packages don't publish 3.9-compatible releases). CI tests 3.10, 3.11, and 3.12 only.
• Known dependency CVEs. pip-audit (run 2026-09-20) found 7 known CVEs in transitive dependencies (werkzeug via flask, nltk via llama-index). See ROADMAP_HONEST.md §4.
• The HTTP servers have no auth layer of their own. They bind to 127.0.0.1 (localhost-only) by default; pass host="0.0.0.0" explicitly only inside a network boundary you control. The event endpoint requires HMAC-SHA256 signatures. See SECURITY.md.

Project status

Beta. Current version 3.3.4 (Development Status :: 4 - Beta). The pure-Python package is the shipped artifact and is under active hardening; the Rust workspace compiles and passes its tests but is not shipped. The most recent pass fixed the Rust build, Python ruff findings, and a stale integration test — see ROADMAP_HONEST.md for the unhedged status.

Installation

pip install PyStreamMCP

Optional extras — install only what you need:

pip install "PyStreamMCP[api]"              # FastAPI + Flask HTTP servers
pip install "PyStreamMCP[mcp]"              # MCP protocol client library
pip install "PyStreamMCP[langchain]"        # LangChain adapter
pip install "PyStreamMCP[llamaindex]"       # LlamaIndex adapter
pip install "PyStreamMCP[semantic-kernel]"  # Semantic Kernel adapter
pip install "PyStreamMCP[all-integrations]" # all supported framework adapters

all-integrations installs only the supported/tested adapters (LangChain, LlamaIndex, Semantic Kernel). There are no extras for experimental modules.

Requires Python 3.9+ for the core SDK; the mcp and semantic-kernel extras require Python 3.10+.

Experimental / unsupported integrations

These modules are not supported, have no optional extra, and are not tested against the real vendor SDKs:

• pystreammcp.integrations.crewai
• pystreammcp.integrations.pydantic_ai
• pystreammcp.integrations.haystack
• pystreammcp.orchestration.temporal
• pystreammcp.orchestration.airflow

They are in-process PyStreamMCP helpers only: they never import the vendor package. Constructing them raises ExperimentalIntegrationError unless you explicitly opt in with experimental=True or PYSTREAMMCP_ENABLE_EXPERIMENTAL=1. TemporalClient is not a Temporal client and the Airflow operators are not BaseOperator subclasses. See docs/INTEGRATIONS.md for the tested support matrix.

Quick start

from pystreammcp import Agent

agent = Agent(
    agent_id="recommendation_engine",
    name="Product Recommendation Engine",
    optimization_strategy="token_efficient",  # or "balanced" / "quality_first"
    max_tokens=1000,
)

result = agent.query("What are the top 10 customers by lifetime value?")

print(f"Baseline tokens:  {result.baseline_tokens}")
print(f"Optimized tokens: {result.optimized_tokens}")
print(f"Reduction:        {result.cost_reduction_percent:.1f}%")

Token counts are estimated from the actual query text (a ~4-characters-per-token heuristic) and the reduction is computed from the selected strategy — not a fixed number. See examples/basic_usage.py for a full walkthrough.

HTTP authentication

Every non-health HTTP route now enforces authentication:

• Set PYSTREAMMCP_API_KEY and send X-API-Key or Authorization: Bearer.
• If unset, only loopback clients are accepted; a network-exposed server fails closed rather than serving queries/sources anonymously.
• /health stays open for probes.

This applies to both the Flask and FastAPI server factories. The MCP tool transport is separate and should be placed behind your MCP host's auth.

Changelog

3.3.4

• Added API-key/Bearer auth to both Flask/FastAPI factories.
• Added token enforcement to MCP call_tool.
• Added DuckDB/SQLGlot federation (pystreammcp[federation]).

Federated execution (DuckDB + SQLGlot)

Cross-source joins are no longer only a not_implemented response. Install the optional backend and register sources:

pip install "pystreammcp[federation]"

from pystreammcp import FederatedQueryEngine

with FederatedQueryEngine() as engine:
    engine.register_sqlite("crm", "crm.db")
    engine.register_parquet("warehouse", "orders.parquet")
    result = engine.execute_join(
        "crm.customers", "warehouse.orders", "l.id = r.customer_id",
        columns="l.id, l.name, r.total",
    )
print(result.columns, result.rows[:5])

Supported sources: SQLite, Parquet and CSV. Queries are read-only, bounded by a row limit, and dialect SQL is transpiled with SQLGlot. An MCP handler created with PyStreamMCPHandler(orchestrator, federation_engine=engine) uses this path for handle_cross_database_join; otherwise the original honest not_implemented behavior remains.

Usage

Data source discovery:

from pystreammcp import SourceRegistry

registry = SourceRegistry()
registry.register(
    "orders_db",
    description="customer order history and revenue by quarter",
    type="database",
    tags=["orders", "revenue"],
)

results = registry.discover("customer revenue by quarter")
# [{"name": "orders_db", "type": "database", "relevance": 0.4444, "matched_terms": [...]}]

MCP orchestration & federation (pystreammcp.toml):

[federation]
endpoints = [
    "http://localhost:8765/mcp",
    "http://localhost:8766/mcp",
]

from pystreammcp import Orchestrator

orch = Orchestrator()  # reads ./pystreammcp.toml
result = orch.discover_mcp_projects()

Webhook event server (HMAC-SHA256):

export PYSTREAMMCP_WEBHOOK_SECRET="a long random shared secret"

Senders compute HMAC-SHA256(secret, raw_request_body) and send it as X-PyStreamMCP-Signature: sha256=<hex digest>. Requests with a missing or invalid signature are rejected with 401; with no secret configured, the endpoint rejects every request with 503.

Testing:

pip install -e ".[dev,api,mcp,langchain,llamaindex,semantic-kernel]"
pytest tests/ -v

tests/test_support_matrix.py asserts that this documented matrix, the extras in pyproject.toml, and the runtime guards stay in sync.

pytest tests/test_support_matrix.py -v

Documentation

• Documentation index
• Cross-repo integrations — wired / real-tested / not tested
• ROADMAP_HONEST.md — full, unhedged status: what's tested, what's fabricated/broken, technical debt, what's explicitly not built
• docs/ARCHITECTURE.md — how the shipped package is structured, with a diagram
• OKF_INTEGRATION.md — the OKF catalog/discovery/query-planner feature
• CHANGELOG.md — release history
• examples/ — runnable examples, including framework integrations and the MCP orchestrator

Known issues and deferred work

Blocked / failed:

• Semantic retrieval/Graphify integration and MCP hardening.

Rejected / deferred:

• The Rust backend; not shipped.

Full shipped/blocked/rejected detail: QUEUE_STATUS.md.

Contributing

Contributions are welcome. See CONTRIBUTING.md and CODE_OF_CONDUCT.md. Report security issues per SECURITY.md.

Related projects

• PyStreamXL — stream large Excel workbooks with constant memory
• PyStreamPDF — token-efficient PDF extraction for RAG
• PyStreamMCP — context and token optimization for AI agents
• PyStreamAI — ML deployment toolkit (canary/A-B, cost tracking, ONNX)
• PyTokenCalc — LLM token counting and cost estimation
• Browse all projects

License

This project is licensed under the Apache License 2.0.