PyStreamMCP 1.0.0

Four-stage selective intelligence pipeline for AI agents - 90-95% token reduction with production observability

PyStreamMCP

The Open Intelligence Layer for AI Agents

PyStreamMCP is the intelligent orchestration layer for LLM systems. Four-stage selective intelligence pipeline with metadata filtering, contextual reranking, quality validation, and production observability. Delivers 90-95% 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?

• 90-95% token reduction without quality loss — Four-stage selective intelligence (metadata filtering → contextual reranking → quality validation → observability)
• Open Knowledge Format (OKF) native — System metadata as git-tracked markdown (no vendor lock-in)
• Production observability — OpenTelemetry tracing, Prometheus metrics, audit trails, decision transparency
• Quality gates built-in — Integrated StatGuardian validation; 10 validation check types; SLA enforcement

Philosophy

Rather than agents asking for everything and parsing the response, PyStreamMCP asks: What does this agent actually need? and Can we deliver it with minimal, highest-value context?

Agent Query
  ↓
STAGE 1: Metadata Filtering (Pre-Retrieval)
  • Rank sources by metadata + quality scores
  • Cache filtering decisions
  • 70-85% reduction
  ↓
Retrieved Content (Minimal Set)
  ↓
STAGE 2: Selective Retrieval (Post-Retrieval)
  • Classify intent + complexity
  • Assign token tiers
  • Rerank by contextual relevance
  • 70-80% reduction
  ↓
Quality-Assured Context
  ↓
STAGE 3: Quality Validation & SLA (Production)
  • Validate source + content quality
  • Enforce SLA compliance
  • Apply fallback strategies
  ↓
STAGE 4: Observability & Tracing (Transparency)
  • OpenTelemetry distributed tracing
  • Prometheus metrics export
  • Audit trail recording
  • 100% decision transparency
  ↓
Optimal Context Window (90-95% reduction)
  ↓
Agent Response (with full transparency + audit trail)

The platform sits between agent frameworks and data systems as a transparent, open intelligence layer—delivering 90-95% token reduction while keeping all decisions traceable, auditable, and production-observable.

Core Capabilities

Stage 1: Metadata Filtering (Pre-Retrieval Intelligence)

• Source Ranking — Rank candidates by metadata (authority, freshness, accessibility, cost, reliability)
• Metadata Caching — Thread-safe async cache with TTL + LRU eviction (1,000 entries max)
• 4 Ranking Strategies — Quality/CostOptimized/Freshness/Balanced
• Query Feature Extraction — Automated analysis of query intent
• Score Justification — Full explanation for every ranking decision
• 70-85% Reduction — Select only top candidates before retrieval

Stage 2: Selective Retrieval (Post-Retrieval Intelligence)

• Intent Classification — Factual/Conceptual/Detailed/Complex query types
• Tiered Token Budgets — Minimal/Standard/Large/Comprehensive with flexible allocation
• Contextual Reranking — 4-dimension scoring (relevance, informativeness, uniqueness, recency)
• Keyword Extraction — Intent-based extraction with stopword removal
• Multiplier Rules — 15+ default rules (critical 2x, financial 1.5x, debug 1.2x, etc.)
• 70-80% Reduction — Filter to token budget with highest-value content

Stage 3: Quality Validation & SLA (Production Assurance)

• 10 Validation Checks — SourceMetadata, Accessibility, Completeness, LanguageMatch, NoPaywall, Freshness, Uniqueness, SignalToNoise, FormatValidity, DataIntegrity
• Pre/Post-Retrieval Validation — Validate before and after fetching
• SLA Enforcement — Quality/latency/availability SLAs with strict/relaxed modes
• 5 Fallback Strategies — RetryWithBackoff/UseAlternativeSource/DegradeGracefully/UseCache/ReturnEmpty
• Confidence Scoring — Low/Medium/High/VeryHigh confidence levels
• 100% Compliance — Quality-assured delivery with fallback recommendations

Stage 4: Observability & Tracing (Production Transparency)

• Distributed Tracing — OpenTelemetry-compatible trace/span management
• Metrics Collection — Per-stage latency (min/max/avg), data reduction %, cache hit rates
• Prometheus Export — Native Prometheus format metrics
• Structured Logging — JSON structured logs with trace ID correlation
• Audit Trail — Complete decision recording with compliance reporting
• 100% Transparency — Full justification for every decision across all 4 stages

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
• Cost Transparency — Every tool has auditable cost, latency, and success profiles

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 (v1.0.0: Four-Stage Pipeline Complete)

Stage 1: Metadata Filtering (Pre-Retrieval)

• Metadata-based source ranking
• Metadata caching (TTL + LRU eviction)
• 4 ranking strategies
• Automatic quality scoring
• 70-85% data reduction

Stage 2: Selective Retrieval (Post-Retrieval)

• Intent classification
• Tiered token budgets
• Contextual reranking (4-dimension scoring)
• 15+ multiplier rules
• 70-80% data reduction

Stage 3: Quality Validation & SLA

• 10 validation check types
• Pre/post-retrieval validation
• SLA enforcement (quality/latency/availability)
• 5 fallback strategies
• Confidence scoring (Low/Medium/High/VeryHigh)

Stage 4: Observability & Tracing

• OpenTelemetry distributed tracing
• Prometheus metrics export
• JSON structured logging
• Audit trail with compliance reporting
• Per-stage decision transparency

Integration & Extensibility

• OKF Native Support — System metadata as git-tracked markdown
• MCP (Model Context Protocol) — Direct integration with Claude and other agents
• Web Knowledge (OSS-only) — SearXNG, Crawl4AI, Trafilatura
• Database Integration — Web/Database/MCPTool metadata types
• OpenTelemetry Compatible — Export to Jaeger, Datadog, NewRelic, Prometheus
• 130+ Tests — Comprehensive test coverage across all 4 stages

Production Ready

• 5,930+ lines of Rust — Type-safe, zero-cost implementation
• < 105ms latency — All 4 stages combined
• 90-95% token reduction — Compound effect across stages
• 100% decision transparency — Full audit trail
• SLA enforcement — Quality gates with fallback strategies

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: Four-Stage Pipeline

PyStreamMCP achieves 90-95% token reduction through compound filtering across four stages:

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 Four-Stage Pipeline):

Stage	Action	Data Before	Data After	Reduction
Stage 1	Rank sources by metadata	115,010	17,252	-85%
Stage 2	Rerank + token budget	17,252	3,450	-80%
Stage 3	Quality validation	3,450	2,070	-40%
Stage 4	Observability metadata	2,070	2,070	~0%
Final	All stages combined	115,010	2,070	-98.2%

• Query: "Which customers are at churn risk?" (10 tokens)
• Top 10 at-risk customers (filtered by Stage 1, reranked by Stage 2): 350 tokens
• Recent interactions (Stage 2 budget allocation): 800 tokens
• Cohort comparison (cached, validated by Stage 3): 400 tokens
• Audit trace (Stage 4 metadata, minimal): 110 tokens
• Total: 1,670 tokens
• Reduction: 98.5% (exceeds 90-95% target)
• Bonus: Full audit trail — Complete decision justification for all 4 stages via Stage 4 tracing

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 — Complete OKF guide + API reference
• CONTRIBUTING.md — Contributing to PyStreamMCP and the OKF catalog
• 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 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.