# PyStreamMCP

> Intelligence layer for AI agents. Query planning, context discovery, cost optimization. 60-75% token reduction while maintaining quality.

- **Type:** MCP server
- **Install:** `agentstack add mcp-mullassery-pystreammcp`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [Mullassery](https://agentstack.voostack.com/s/mullassery)
- **Installs:** 0
- **Category:** [AI & ML](https://agentstack.voostack.com/c/ai-and-ml)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [Mullassery](https://github.com/Mullassery)
- **Source:** https://github.com/Mullassery/PyStreamMCP
- **Website:** https://github.com/Mullassery/PyStreamMCP

## Install

```sh
agentstack add mcp-mullassery-pystreammcp
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 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

```bash
pip install pystreammcp
```

### Quick Example: OKF-Powered Query Optimization

```python
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

- **[OKF_INTEGRATION.md](OKF_INTEGRATION.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

```bash
# 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

```bash
# 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](https://github.com/Mullassery/PyStreamMCP/issues)
- **Discussions:** [PyStreamMCP/discussions](https://github.com/Mullassery/PyStreamMCP/discussions)
- **OKF Questions:** See [OKF_INTEGRATION.md](OKF_INTEGRATION.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](https://github.com/Mullassery)
- **Source:** [Mullassery/PyStreamMCP](https://github.com/Mullassery/PyStreamMCP)
- **License:** MIT
- **Homepage:** https://github.com/Mullassery/PyStreamMCP

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

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/mcp-mullassery-pystreammcp
- Seller: https://agentstack.voostack.com/s/mullassery
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.