Model Context Protocol (MCP)

Koveria's Model Context Protocol (MCP) is the foundational system that transforms AI agents from simple LLM wrappers into full-featured autonomous systems capable of real-world work.

What is MCP?

MCP is a standardized protocol for extending AI agent capabilities through service integrations. Think of MCPs as "apps" for your AI agents - each MCP service provides specific functionality that agents can leverage.

The Problem MCP Solves

Without MCP:

# Agent can only talk to LLMs
response = await self.llm.generate("What's the weather in Berlin?")
# Result: "I don't have access to real-time weather data..."

With MCP:

# Agent can call external APIs
weather = await self.mcp.http_client.get("https://api.weather.com/berlin")
# Agent can process structured data
data = await self.mcp.json_yaml.parse_json(weather.text)
# Agent can perform calculations
temp_f = await self.mcp.math.convert_celsius_to_fahrenheit(data["temp_c"])
# Result: Real, actionable weather information

---

Quick Example: The Power of MCP

Scenario: An AI agent needs to process customer invoices.

Traditional Approach (No MCP)

# ❌ Limited capabilities
async def process_invoice(self, invoice_text: str):
    # Agent can only analyze text via LLM
    response = await self.llm.generate(
        f"Analyze this invoice: {invoice_text}"
    )
    # Can't:
    # - Read PDF files
    # - Extract structured data
    # - Validate totals
    # - Send confirmation emails
    # - Store results

MCP-Powered Approach

# ✅ Full automation
async def process_invoice(self, invoice_path: str):
    # 1. Read PDF invoice
    pdf_text = await self.mcp.pdf_processing.extract_text(
        pdf_path=invoice_path
    )

    # 2. Parse invoice data
    invoice_data = await self.mcp.json_yaml.parse_json(pdf_text)

    # 3. Validate calculations
    subtotal = await self.mcp.math.sum(invoice_data["line_items"])
    tax = await self.mcp.finance.calculate_tax(subtotal, rate=0.19)
    total = subtotal + tax

    # 4. Validate against invoice
    is_valid = await self.mcp.validation.validate_number(
        invoice_data["total"],
        equals=total,
        tolerance=0.01
    )

    # 5. Send confirmation email
    if is_valid:
        await self.mcp.email_send.send_email(
            to=invoice_data["customer_email"],
            subject=f"Invoice {invoice_data['number']} Processed",
            body=f"Your invoice for {total}€ has been approved."
        )

    # 6. Store in cloud storage
    await self.mcp.cloud_storage.write_file(
        path=f"/invoices/processed/{invoice_data['number']}.json",
        content=json.dumps(invoice_data)
    )

Result: The agent can now fully automate invoice processing - something impossible with LLMs alone.

---

MCP Architecture Overview

---

MCP Service Classification

Koveria uses a three-dimensional classification to organize MCPs:

1. Execution Model (How it runs)

Execution	Latency	Use Cases	Examples
Local	<1ms	Lightweight operations, utilities	JSON parsing, date formatting, math
Remote	5-50ms	Heavy processing, external APIs	PDF extraction, web scraping, email

2. Service Category (What it provides)

Category	Description	Examples
stdlib	Standard library utilities (local)	JSON, DateTime, Math, Validation
platform	Core infrastructure services (remote)	Cloud Storage, HTTP Client, Email
integration	Third-party connectors	Stripe, Salesforce, Slack
custom	Organization-specific	Your custom business logic

3. Ownership (Who maintains it)

Ownership	Provider	Examples
core	Koveria (built-in)	All stdlib + platform MCPs
community	Community marketplace	Popular integrations
org	Your organization (private)	Internal tools

Example Classification:

• JSON/YAML MCP = Core Standard Library (Local)
• HTTP Client MCP = Core Platform Service (Remote)
• Stripe MCP = Community Integration (Remote)

---

The 13 Core MCP Services

Koveria includes 13 foundational MCP services out of the box:

Standard Library MCPs (Local, <1ms)

1. JSON/YAML MCP - Parse, validate, query JSON/YAML
2. DateTime MCP - Date/time parsing, formatting, timezones
3. Math MCP - Mathematical operations, statistics
4. Text Processing MCP - String manipulation, token counting
5. Validation MCP - Email, URL, phone validation
6. Finance MCP - Financial calculations, currency conversion
7. Hashing MCP - Cryptographic hashing
8. Encoding MCP - Base64, URL encoding
9. Business Date MCP - Business day calculations

Platform MCPs (Remote, 5-50ms)

10. Cloud Storage MCP - S3/MinIO file operations
11. HTTP Client MCP - HTTP API calls
12. Email Send MCP - SMTP email sending
13. PDF Processing MCP - PDF text/table extraction

See: Standard Library Overview | Platform Services Overview

---

Configuration Hierarchy

MCPs support multi-level configuration for maximum flexibility:

Platform Defaults (least specific)
    ↓
Organization Settings
    ↓
Workspace Settings
    ↓
Team Settings
    ↓
User Settings (most specific)

Example: HTTP Client MCP Configuration

# Organization-wide (Admin GUI)
org_mcp_config:
  http_client:
    allowed_domains:
      - "*.example.com"
      - "api.stripe.com"
    max_request_size_mb: 10
    timeout_seconds: 30

# Team-level override (Customer Portal)
team_mcp_config:
  http_client:
    allowed_domains:
      - "api.internal-tool.com"  # Additional domain for this team

Benefits:

• Security by default - Org admin sets boundaries
• Team flexibility - Teams customize within limits
• Cost attribution - Track usage per team
• Audit trail - All config changes logged

See: Configuration Hierarchy

---

Building Custom MCPs

Koveria provides a developer-first SDK for building custom MCP services with zero boilerplate.

Quick Example: Stripe Payment MCP

from koveria.mcp import MCPService, mcp_tool, mcp_config
from koveria.mcp.types import MCPExecutionModel, MCPCategory, MCPOwnership

@mcp_config.api_key(
    name="stripe_api_key",
    description="Stripe API key",
    secret=True  # Stored in Vault
)
@mcp_config.select(
    name="environment",
    options=["test", "live"],
    default="test"
)
class StripeMCP(MCPService):
    """Stripe payment integration."""

    name = "stripe"
    display_name = "Stripe Payments"
    version = "1.0.0"

    # Classification
    __execution__ = MCPExecutionModel.REMOTE
    __category__ = MCPCategory.INTEGRATION
    __ownership__ = MCPOwnership.ORG

    @mcp_tool(description="Create a payment intent")
    async def create_payment(
        self,
        amount: int,
        currency: str = "usd",
        customer_id: str = None
    ) -> dict:
        """Create a Stripe payment intent."""
        return await self.stripe.PaymentIntent.create(
            amount=amount,
            currency=currency,
            customer=customer_id
        )

That's it! The platform automatically:

• ✅ Generates Admin GUI configuration form
• ✅ Handles secret storage in Vault
• ✅ Provides koveria mcp dev local development server
• ✅ Builds Docker image with koveria mcp build
• ✅ Deploys to registry with auto-discovery
• ✅ Enables agents to call await self.mcp.stripe.create_payment()

See: Build Your First MCP

---

MCP Development Workflow

Key Features:

• Instant feedback - koveria mcp dev for rapid iteration
• Zero config - Decorators define everything
• Auto-discovery - Push to registry, platform finds it
• Type safety - Full Python type hints

See: MCP Developer Guide

---

Use Cases

1. Compliance Automation (70% Effort Reduction)

Problem: Compliance assessments require gathering evidence from documentation, code, and APIs.

Solution: Koveria's PlaceholderResolver agent uses MCPs to automate evidence collection:

# Compliance agent uses MCPs to prefill questionnaires
async def prefill_questionnaire(self, agent_id: str):
    # 1. Read agent documentation
    readme = await self.mcp.cloud_storage.read_file(
        path=f"/agents/{agent_id}/README.md"
    )

    # 2. Scan code for patterns
    code_files = await self.mcp.cloud_storage.list_directory(
        path=f"/agents/{agent_id}/src",
        pattern="*.py"
    )

    # 3. Extract metadata from config
    config = await self.mcp.json_yaml.parse_yaml(
        yaml_string=await self.mcp.cloud_storage.read_file(
            path=f"/agents/{agent_id}/koveria.yaml"
        )
    )

    # 4. Use LLM to infer answers from evidence
    answers = await self.llm.generate(
        f"Based on: {readme}, {code_files}, {config}..."
    )

    # Result: 70% of compliance questions automatically answered

2. Invoice Processing

Problem: Manual invoice processing is slow and error-prone.

Solution: AI agent with PDF, JSON, Math, Email MCPs (see Quick Example above).

3. Customer Support Automation

Problem: Support agents need to check multiple systems (CRM, ticketing, knowledge base).

Solution: AI agent with HTTP Client, JSON, Text Processing MCPs:

async def handle_support_ticket(self, ticket_id: str):
    # 1. Fetch ticket from Zendesk API
    ticket = await self.mcp.http_client.get(
        url=f"https://api.zendesk.com/tickets/{ticket_id}"
    )

    # 2. Parse ticket data
    data = await self.mcp.json_yaml.parse_json(ticket.text)

    # 3. Search knowledge base
    kb_results = await self.mcp.http_client.post(
        url="https://kb.internal.com/search",
        body={"query": data["subject"]}
    )

    # 4. Generate response with LLM
    response = await self.llm.generate(
        f"Generate support response based on: {kb_results}"
    )

    # 5. Update ticket via API
    await self.mcp.http_client.patch(
        url=f"https://api.zendesk.com/tickets/{ticket_id}",
        body={"comment": response}
    )

4. Financial Report Generation

Problem: Monthly financial reports require data from multiple sources, calculations, and formatting.

Solution: AI agent with Cloud Storage, CSV/Excel, Finance, Math MCPs:

async def generate_monthly_report(self, month: str):
    # 1. Read transaction CSV
    transactions = await self.mcp.csv_excel.read_csv(
        file_path=f"/finance/transactions-{month}.csv"
    )

    # 2. Calculate totals by category
    by_category = {}
    for tx in transactions:
        category = tx["category"]
        amount = float(tx["amount"])
        by_category[category] = await self.mcp.math.sum([
            by_category.get(category, 0),
            amount
        ])

    # 3. Calculate tax
    total = await self.mcp.math.sum(by_category.values())
    tax = await self.mcp.finance.calculate_tax(total, rate=0.19)

    # 4. Generate Excel report
    await self.mcp.csv_excel.write_excel(
        rows=[
            {"Category": k, "Amount": v}
            for k, v in by_category.items()
        ],
        file_path=f"/finance/reports/monthly-{month}.xlsx",
        sheet_name="Summary"
    )

---

Security & Compliance

Multi-Layer Security

1. Row-Level Security (RLS)

• All MCP configurations scoped by org_id
• Database-enforced tenant isolation
• Zero-trust model

2. Secrets Management

• API keys stored in Vault (never in code)
• Automatic rotation policies
• Audit trail for all secret access

3. Rate Limiting

• Per-org, per-workspace, per-team quotas
• Prevent abuse and cost overruns
• Configurable limits

4. Audit Logging

• All MCP calls logged with metadata
• 7-year retention for compliance
• Immutable audit trail

5. Network Isolation

• Teams cannot access each other's MCP data
• NetworkPolicy enforcement in Kubernetes
• mTLS between services

See: MCP Security Model

Cost Attribution

Every MCP call is tracked for cost attribution:

{
  "timestamp": "2026-01-03T10:30:00Z",
  "org_id": "org-001",
  "team_id": "team-sales",
  "agent_id": "agent-invoice-processor",
  "mcp_name": "pdf_processing",
  "mcp_tool": "extract_text",
  "cost_usd": 0.05,
  "duration_ms": 234
}

Benefits:

• Chargeback - Bill teams for MCP usage
• Budgets - Set team-level spending limits
• Optimization - Identify expensive operations

See: Cost Attribution

---

Getting Started

For Developers: Build Your First MCP

Time: 10 minutes

# 1. Initialize new MCP project
koveria mcp init weather-api

# 2. Edit src/main.py (auto-generated scaffold)
# Add @mcp_config and @mcp_tool decorators

# 3. Start local dev server
koveria mcp dev

# 4. Test from agent
await self.mcp.weather_api.get_forecast(city="Berlin")

See: Build Your First MCP

For Admins: Configure Org-Wide MCPs

Time: 5 minutes

1. Navigate to Admin GUI → MCP Configuration
2. Select MCP (e.g., HTTP Client)
3. Configure settings:
   • Allowed domains
   • Rate limits
   • Timeout values
4. Save → Configuration applies to all teams

See: MCP Admin Configuration

For Agent Developers: Use MCPs in Agents

Time: 2 minutes

from koveria import Agent

class InvoiceProcessorAgent(Agent):
    async def process_invoice(self, pdf_path: str):
        # MCPs available via self.mcp.*
        text = await self.mcp.pdf_processing.extract_text(
            pdf_path=pdf_path
        )

        data = await self.mcp.json_yaml.parse_json(text)

        return data

See: SDK Quick Start

---

Documentation Navigation

Core Concepts

• What is MCP? - Conceptual introduction
• Execution Models - Local vs Remote
• Classification System - 3D classification
• Manifest Files - koveria.yaml schema
• Configuration Hierarchy - Multi-level config

Service Catalogs

• Standard Library MCPs - 9 local services (<1ms)
• Platform MCPs - 4 remote services (5-50ms)

Developer Guides

• Build Your First MCP - Step-by-step tutorial
• MCP SDK Overview - SDK introduction
• Configuration Decorators - @mcp_config guide
• Tool Decorators - @mcp_tool guide

Admin Guides

• Org-Wide Configuration - Configure MCPs
• Team Assignment - Assign MCPs to teams
• Secrets Management - Vault integration

Reference

• CLI Commands - koveria mcp CLI
• SDK API - Python SDK reference
• Manifest Schema - koveria.yaml schema

---

FAQs

What's the difference between local and remote MCPs?

Local MCPs run in-process with your agent (as Python libraries):

• Latency: <1ms
• Use cases: Lightweight operations (JSON parsing, math, validation)
• Examples: JSON/YAML, DateTime, Math

Remote MCPs run as separate Kubernetes pods:

• Latency: 5-50ms (network hop)
• Use cases: Heavy processing, external APIs
• Examples: PDF processing, HTTP Client, Email

See: Execution Models

Can I use MCPs from multiple teams?

Yes! MCP servers are org-wide by default, but access is controlled at the team level:

1. Admin configures MCP org-wide (e.g., HTTP Client)
2. Teams enable specific MCPs for their use
3. Cost is attributed to the team that uses it

See: Team Assignment

How do I secure API keys?

All secrets are stored in Vault, never in code:

@mcp_config.api_key(
    name="stripe_api_key",
    secret=True  # ← Stored in Vault automatically
)
class StripeMCP(MCPService):
    # Koveria handles secret injection
    pass

See: Secrets Management

What happens if an MCP call fails?

MCPs return structured errors:

try:
    result = await self.mcp.http_client.get("https://api.example.com")
except MCPError as e:
    # Error details
    print(e.mcp_name)  # "http_client"
    print(e.tool_name)  # "http_get"
    print(e.error_code)  # "TIMEOUT"
    print(e.message)  # "Request timed out after 30s"

All errors are logged to audit trail for compliance.

---

Next Steps

1. What is MCP? - Deep dive into MCP concepts
2. Build Your First MCP - 10-min tutorial
3. Standard Library MCPs - Explore built-in services
4. Platform MCPs - Explore infrastructure services

---

Last Updated: January 3, 2026 Version: 1.0.0 (Architecture v9.0)