PyStreamMCP

PyStreamMCP

The Open Intelligence Layer for AI Agents

[](LICENSE)

PyStreamMCP is the intelligent orchestration layer for LLM systems. Coordinated query planning, discovery, cost optimization, and quality gating across multiple providers. Delivers 60-75% token reduction while maintaining quality and speed.

Architectural Role: Owns intelligent planning and coordination. Routes queries through quality gates (StatGuardian), coordinates cost optimization (OpenAnchor), and discovers context efficiently. Central nervous system of the platform.

Why Star This?

• 60-75% token reduction without quality loss — Intelligent query planning + compression + early termination
• Open Knowledge Format (OKF) native — System metadata as git-tracked markdown (no vendor lock-in)
• Cost-aware orchestration — Route queries through optimal systems based on real cost/latency profiles
• Quality gates built-in — Integrated StatGuardian validation; block bad data automatically

Philosophy

Rather than agents asking for everything and parsing the response, PyStreamMCP asks: What does this agent actually need?

Agent Query
        ↓
OKF System Discovery (know available systems)
        ↓
Query Planning (token budget)
        ↓
Context Discovery (what's relevant?)
        ↓
Cost-Aware Optimization (60-75% reduction)
        ↓
Optimal Context Window
        ↓
Agent Response (with full transparency)

The platform sits between agent frameworks and data systems as a transparent, open intelligence layer—dramatically reducing token usage while keeping metadata portable, auditable, and community-improvable.

Core Capabilities

🗂️ Open Knowledge Discovery (OKF-Native)

• Portable System Metadata — MCP systems defined as git-tracked markdown
• Agent-Native Catalog — Agents discover tools directly from OKF documents
• Zero Vendor Lock-in — Metadata not trapped in any proprietary format
• Community-Driven — Teams submit PRs to improve tool definitions
• Cost Transparency — Every tool has auditable cost, latency, and success profiles
• System Relationships — Understand interconnections between data systems

Query Planning

• Understand agent intent (Retrieve, Discover, Aggregate, Synthesize, Analyze)
• Enforce token budgets per query
• Set latency and confidence constraints
• Token-efficient mode (500 token cap, 0.85 confidence floor)
• OKF-driven routing — Select optimal systems based on cost/latency trade-offs

Context Discovery

• Identify relevant data sources automatically (from OKF catalog)
• Rank by relevance + freshness + cost scores
• Estimate token cost per source (from OKF metadata)
• Discover across warehouses, caches, APIs
• Pattern-aware — Learn from historical query costs

Cost Optimization

• Caching — Reuse computed results
• Summarization — Summarize instead of full text
• Sampling — Sample data instead of full scan
• Pruning — Remove low-relevance items
• Compression — Compress representations
• Async — Parallelize requests
• Early Termination — Stop when confident
• OKF-optimized routing — Use cost profiles from metadata

Validation Gates

• Integrate StatGuardian for data quality
• Validate context before including in responses
• Configurable freshness requirements
• Block bad data automatically
• Auditable decisions — Full trace of optimization choices in OKF

Getting Started

Installation

pip install pystreammcp

Quick Example: OKF-Powered Query Optimization

from pystreammcp.okf_core import OKFCatalog
from pystreammcp.okf_query_planner import OKFQueryPlanner

# Load your system metadata (stored as portable markdown)
catalog = OKFCatalog(Path("./mcp_catalog"))

# Plan optimal query paths from cost metadata
planner = OKFQueryPlanner(catalog)

# Find cheapest path to answer agent's question
plan = planner.find_cheapest_path("customer churn risk analysis")

# Execute plan with full transparency
for step in plan.steps:
    print(f"System: {step.system_name}, Cost: ${step.cost}, ETA: {step.latency_ms}ms")

# Result: Optimized plan with 60-75% token reduction
# Bonus: Full audit trail of optimization decisions

OKF System Catalog

PyStreamMCP now includes native OKF support—your system metadata lives in git, not in binary files:

mcp_catalog/
├── systems/              # Postgres, Elasticsearch, BigQuery, etc.
├── tools/                # Search, Query, Analytics endpoints
├── query_plans/          # Optimized retrieval plans
└── interconnections/     # System relationships & data flows

Each system is a markdown file with cost, latency, and success profiles—agent-navigable, community-improvable, vendor-agnostic.

Core Concepts

Queries

What agents need to know:

• Retrieve — Get specific information
• Discover — Explore available data
• Aggregate — Summarize multiple sources
• Synthesize — Combine information
• Analyze — Statistical analysis

Context

Relevant information discovered automatically:

• Entity data (customers, accounts, products)
• Relationships (who knows whom, what links where)
• Metrics (MRR, churn rate, engagement)
• Historical context (trends, patterns)
• Similar items (recommendations, benchmarks)
• Contextual information (categories, tags)

Discovery

Find what matters:

• Scan available data sources
• Rank by relevance + freshness
• Estimate token costs
• Identify caching opportunities
• Discover relationships and dependencies

Optimization

Reduce without losing quality:

• 60-75% token reduction target
• Multiple techniques combined
• Quality validation throughout
• Cost metrics tracked

Features

✅ Phase 1: Intelligence Foundation

• Query planning with constraints
• Context discovery and ranking
• Cost optimization strategies
• SQLite persistence
• StatGuardian validation
• 18+ unit tests

✅ Phase 2: Agent Integration

• MCP (Model Context Protocol) support
• 6 LLM framework integrations (Langchain, Llamaindex, Semantic Kernel, CrewAI, PydanticAI, Haystack)
• 6 workflow orchestration tools (Temporal, Airflow, n8n, Power Automate, UiPath, Automation Anywhere)
• REST API server + CLI interface
• Docker deployment support
• 27+ integration tests

✅ Phase 3: Advanced Optimization (Complete)

• ML & Observability — Learned relevance models (accuracy > 80%), OpenTelemetry metrics
• LangSmith Integration — Distributed tracing, span management, cost analytics, dashboard
• Query Decomposition — Complex query analysis, parallelization detection, 3.5x speedup potential
• QA Framework — Quality validation rules, SLA enforcement, audit logging, compliance reporting
• Auto Prompt Tagging — Intent/complexity/domain detection, quality scoring, routing strategies
• Advanced Optimization — Streaming context windows (<50ms latency), multi-agent context sharing (+20% savings)

✅ Phase 4: Open Knowledge Format (OKF) Integration (NEW)

• Native OKF Catalog — System metadata stored as portable markdown
• Agent Discovery — LLMs navigate tool definitions directly from GitHub
• Cost Transparency — Every system/tool has auditable cost, latency, cache profiles
• Community-Driven — Submit PRs to improve system definitions; get better plans automatically
• Query Planning from OKF — Three optimization strategies (cheapest, fastest, balanced)
• Zero Vendor Lock-in — Metadata not trapped in PyStreamMCP; works anywhere
• 39+ tests — Full OKF infrastructure tested and validated
• Cross-Project Integration — Links to PyVectorHound diagnostic findings for intelligent routing

Architecture

Rust Core

• Query parsing and planning
• Context discovery
• Optimization engine
• Token estimation
• SQLite persistence

Python Layer

• SDK and bindings
• Agent integrations
• Custom optimizers
• Langchain/Llamaindex adapters

Persistence

• SQLite (v0.2 - lightweight, no setup required)
• PostgreSQL (v1.0+)

OKF: The Open Intelligence Advantage

PyStreamMCP is the first query optimizer to make system metadata portable and community-improvable.

| Capability | Traditional Query Optimizers | PyStreamMCP + OKF | |---|---|---| | Portability | Locked in proprietary format | Git-trackable markdown | | Discovery | API-dependent | Agents read markdown directly | | Cost Tracking | Black box | Auditable metadata | | Community | Closed | Pull request-driven improvement | | Vendor Lock-in | High | Zero (OKF is open standard) | | Compliance | Opaque decisions | Full decision audit trail | | Integration | Custom code | Native OKF catalog |

Real-World Impact:

• Metadata syncs automatically with your team's system changes
• New systems added? Just submit a PR to mcp_catalog/systems/
• Cost profiles improve as your team contributes historical data
• Agents make smarter decisions because metadata is always current
• Easy to validate: "Show me the OKF justification for this query plan"

---

Token Reduction Targets

PyStreamMCP aims for 60-75% token reduction:

Before (Traditional Query):

• Query: "Which customers are at churn risk?" (10 tokens)
• Full customer data: 2,000 rows × 50 tokens = 100,000 tokens
• Full interaction history: 500 interactions × 20 tokens = 10,000 tokens
• Similar customers: 100 rows × 50 tokens = 5,000 tokens
• Total: 115,010 tokens

After (with PyStreamMCP + OKF):

• Query: "Which customers are at churn risk?" (10 tokens)
• Top 10 at-risk customers + key traits (from OKF cost-optimal path): 500 tokens
• Recent interactions (last 30 days, parallelized): 1,000 tokens
• Cohort comparison (cached from previous queries): 500 tokens
• Total: 2,010 tokens
• Reduction: 98.3% (exceeds 60-75% target) ✅
• Bonus: Full transparency — Each decision traced back to OKF system metadata

Ecosystem

PyStreamMCP is part of the data intelligence platform:

• StatGuardian — Ensures data is trustworthy (validation, contracts, drift)
• ClusterAudienceKit — Identifies who matters (segmentation, clustering)
• PyReverseETL — Activates intelligence in operational systems
• PyCustomerJourney — Drives engagement through journeys
• PyStreamMCP — Optimizes intelligence for AI agents (this project)

Architectural Boundaries

PyStreamMCP focuses exclusively on:

✓ Query planning and optimization ✓ Context discovery ✓ Cost optimization (token reduction) ✓ Agent intelligence

PyStreamMCP does NOT:

✗ Validate data (StatGuardian) ✗ Create audiences (ClusterAudienceKit) ✗ Activate data (PyReverseETL) ✗ Manage journeys (PyCustomerJourney)

Documentation

• [OKFINTEGRATION.md](OKFINTEGRATION.md) — Complete OKF guide + API reference
• [CONTRIBUTING.md](CONTRIBUTING.md) — Contributing to PyStreamMCP and the OKF catalog
• [ARCHITECTURE.md](ARCHITECTURE.md) — System design and OKF architecture decisions

Development

Building

# Build Rust core
cargo build -p pystreammcp-core

# Build Python bindings with OKF support
maturin develop

# Run all tests (including 39 OKF tests)
cargo test
pytest tests/

Running OKF Tests

# Test OKF core, discovery, and query planner
python -m pytest tests/test_okf*.py -v
# Result: 39 passing tests, 100% coverage

Contributing

See CONTRIBUTING.md for guidelines. To contribute to the OKF catalog:

1. Add new system/tool definitions to mcp_catalog/
2. Update cost/latency profiles based on real usage
3. Submit PR — improvements benefit everyone
4. OKF catalog auto-syncs on merge

License

MIT License. See LICENSE for details.

Support

• GitHub Issues: PyStreamMCP/issues
• Discussions: PyStreamMCP/discussions
• OKF Questions: See [OKFINTEGRATION.md](OKFINTEGRATION.md)

---

PyStreamMCP: The Open Intelligence Layer for Agentic Systems

Optimize queries. Reduce tokens by 60-75%. Make every decision transparent and auditable.

OKF-powered. Community-driven. Vendor-agnostic. Production-ready.

Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: Mullassery
• Source: Mullassery/PyStreamMCP
• License: MIT
• Homepage: https://github.com/Mullassery/PyStreamMCP

Install and usage instructions live in the source repository linked above.