appmcp 1.0.1

Embed curated MCP tools, resources, and prompts in an existing Python application.

AppMCP

Add a secure, embedded MCP server to any Python application.

Package name: appmcp
Project name: AppMCP
Install: pip install appmcp
Import: from appmcp import AppMCP

AppMCP embeds a curated Model Context Protocol server inside an existing Python application without duplicating business logic or starting a second process.

AppMCP combines:

• Embedded Streamable HTTP powered by FastMCP.
• Explicit capability exposure with @tool, @resource, and @prompt.
• Existing service registration with optional tool namespaces.
• Application dependency injection for objects, sync factories, and async factories.
• Request context containing the user, headers, request ID, app, and services.
• Security policies for authentication, scopes, confirmation, availability, and rate limits.
• FastAPI, Starlette, and generic ASGI mounting with coordinated lifespans.
• Middleware and observability through audit sinks and OpenTelemetry spans.
• In-memory testing without opening a port.
• Backend isolation so public APIs do not depend directly on FastMCP.

AppMCP is a safe application integration layer, not an automatic REST-to-MCP converter. Only explicitly decorated capabilities are exposed.

---

Before / After

Without AppMCP, applications often duplicate business logic in a separate MCP project and run another server process:

REST endpoint -> OrderService -> Repository
MCP tool      -> duplicated order lookup -> Repository

With AppMCP, the same service method remains usable by REST endpoints, background workers, tests, and MCP clients:

from appmcp import AppMCP, tool


class OrderService:
    def __init__(self, repository):
        self.repository = repository

    @tool(name="find", read_only=True, scopes=["orders:read"])
    async def find_order(self, order_id: str) -> dict:
        return await self.repository.find(order_id)


mcp = AppMCP(name="Store", default_deny=True)
mcp.include(OrderService(repository), namespace="orders")
mcp.mount(app, path="/mcp")

The application now exposes orders.find, while find_order() remains a normal Python method and every undecorated method stays private.

---

Why Embed MCP in the Application?

Separate MCP services introduce another deployment, lifecycle, authentication boundary, and implementation of the same operations:

Existing application
  - services
  - repositories
  - authentication
  - lifecycle
    |
    v
AppMCP integration
  - explicit capability registry
  - dependency and context injection
  - application security policies
  - FastMCP transport
    |
    v
POST /mcp in the same process

AppMCP is designed to avoid:

• Duplicated business logic between REST handlers and MCP tools.
• Accidental exposure from automatically publishing every public method.
• Separate dependency graphs for databases, repositories, and services.
• Split authentication state between the application and MCP server.
• Extra deployment complexity for applications that only need one process.
• Network-heavy tests when deterministic in-memory calls are sufficient.

---

Why Not Generate Tools from Every REST Endpoint?

OpenAPI conversion is useful for broad API coverage, but complex REST APIs do not always produce focused tools for agents. AppMCP favors deliberately designed capabilities with clear names, schemas, descriptions, and security policies.

That makes AppMCP a curated integration layer rather than an automatic API mirror.

---

Architecture

Existing Python application
  - FastAPI / Starlette / ASGI
  - services and repositories
  - authentication and application state
    |
    v
AppMCP
  - decorators and explicit registry
  - dependency container
  - request context
  - scopes, confirmation, availability, rate limits
  - middleware, audit, telemetry
    |
    v
MCPBackend protocol
  - FastMCP backend
  - official SDK compatibility backend
  - custom application backend
    |
    v
Streamable HTTP at /mcp

The public API depends on the MCPBackend protocol. FastMCP is the default engine and owns the MCP protocol implementation.

---

Install

pip install appmcp

For FastAPI development with Uvicorn:

pip install "appmcp[fastapi]"

For Redis-backed sessions and coordinated rate limits:

pip install "appmcp[redis]"

For OpenTelemetry support:

pip install "appmcp[otel]"

Experimental framework and backend integrations:

pip install "appmcp[flask]"
pip install "appmcp[django]"
pip install "appmcp[official-sdk]"

For development:

pip install -e ".[dev,fastapi]"
pytest -q
python -m build

---

Quick Start

Embed AppMCP in FastAPI

from fastapi import FastAPI
from appmcp import AppMCP, MCPContext

app = FastAPI(title="Orders API")
mcp = AppMCP(app, name="Orders MCP", path="/mcp")


@app.get("/health")
async def health() -> dict[str, bool]:
    return {"ok": True}


@mcp.tool(read_only=True)
async def get_order(order_id: str, ctx: MCPContext) -> dict:
    return {
        "id": order_id,
        "request_id": ctx.request_id,
        "status": "processing",
    }

Run the application normally:

uvicorn myapp:app --reload

The same process now serves both application and MCP traffic:

GET  /health
GET  /api/...
POST /mcp

Register Existing Services

from appmcp import tool


class PaymentService:
    @tool(name="refund", scopes=["payments:write"], confirmation_required=True)
    async def refund_payment(self, payment_id: str) -> dict:
        return await self.gateway.refund(payment_id)


mcp.include(PaymentService(), namespace="payments")

The resulting MCP tool is payments.refund.

Inject Application Dependencies

mcp.provide("documents", document_service)


@mcp.tool()
async def search(
    query: str,
    documents=mcp.depends("documents"),
) -> list[dict]:
    return await documents.search(query)

Providers may be objects, synchronous factories, or asynchronous factories. A factory may accept ctx or an MCPContext parameter.

Use Request Context

from appmcp import MCPContext


@mcp.tool(scopes=["orders:read"], read_only=True)
async def my_orders(ctx: MCPContext) -> list[dict]:
    ctx.logger.info("Listing orders for request %s", ctx.request_id)
    return await orders.for_user(ctx.user.id)

Context includes user, headers, request_id, session, app, services, logger, scopes, confirmation state, and custom metadata.

---

Security

Bearer Authentication

from appmcp import AppMCP, BearerAuth, Identity


async def validate_token(token: str) -> Identity | None:
    account = await accounts.from_token(token)
    if account is None:
        return None
    return Identity(account, frozenset(account.scopes))


mcp = AppMCP(
    app,
    name="Store",
    auth=BearerAuth(validate_token),
    default_deny=True,
    expose_errors=False,
)

AppMCP does not validate JWT signatures or OAuth claims itself. The application callback must verify signatures, issuer, audience, expiry, and revocation.

Capability Policies

@mcp.tool(
    scopes=["orders:write"],
    rate_limit="10/minute",
    confirmation_required=True,
)
async def cancel_order(order_id: str) -> dict:
    return await orders.cancel(order_id)

Dynamic availability can reject capabilities based on the current context:

@mcp.tool(enabled=lambda ctx: ctx.user.plan == "enterprise")
async def export_all_orders() -> dict:
    return await orders.export_all()

The default limiter is process-local. Use RedisRateLimiter for coordinated multi-worker enforcement.

---

Resources and Prompts

@mcp.resource("orders://summary")
def order_summary() -> dict:
    return {"open": 12, "processing": 4}


@mcp.prompt()
def review_order(order_id: str) -> str:
    return f"Review order {order_id} for fulfilment risks."

Resources and prompts support explicit registration, descriptions, scopes, namespaces, and dynamic availability.

---

Middleware and Observability

Middleware receives the capability definition, request context, arguments, and the next handler:

async def timing_middleware(definition, context, arguments, call_next):
    context.logger.info("Calling %s", definition.name)
    return await call_next()


mcp.use(timing_middleware)

AuditLogger accepts synchronous or asynchronous sinks. OpenTelemetry spans are emitted automatically when the optional API is installed.

Audit events may contain arguments and user objects. Production sinks should redact secrets and personal data before storage.

---

Testing Without a Port

async def test_find_order():
    async with mcp.test_client(
        user=user,
        scopes=["orders:read"],
    ) as client:
        result = await client.call_tool(
            "orders.find",
            {"order_id": "order-123"},
        )

        assert result.data["id"] == "order-123"

The test client runs policies, dependencies, middleware, telemetry, and audit hooks directly. Add a FastMCP client integration test when transport behavior is part of the contract under test.

---

CLI

Inspect exposed capabilities:

appmcp inspect myapp.main:app

Run startup-grade validation and production diagnostics:

appmcp validate myapp.main:app
appmcp doctor myapp.main:app

Validate registration:

appmcp test myapp.main:app

Run the ASGI development server:

appmcp dev myapp.main:app --port 8000

Call a tool and manage schema snapshots:

appmcp call myapp.main:app orders.find --arg order_id=123
appmcp schema myapp.main:app --output appmcp-schema.json
appmcp schema --diff previous.json appmcp-schema.json
appmcp init

Generate client configuration:

appmcp config claude myapp.main:app --url http://127.0.0.1:8000/mcp
appmcp config cursor myapp.main:app --url http://127.0.0.1:8000/mcp
appmcp config vscode myapp.main:app --url http://127.0.0.1:8000/mcp

inspect and test validate the registry. They do not replace the application test suite or make live tool calls.

---

Main Features

1. Embedded MCP Server

Mount Streamable HTTP in an existing FastAPI, Starlette, or generic ASGI application. AppMCP composes mounted lifespans so the MCP backend starts and stops with the host application.

2. Explicit Registration

Only methods marked with @tool, @resource, or @prompt are scanned. Public methods are never exposed automatically.

3. Tool Namespaces

mcp.include(order_service, namespace="orders")
mcp.include(payment_service, namespace="payments")

This produces focused names such as orders.find and payments.refund.

4. Dependency Container

Use the application's existing service objects and factories without creating a second dependency graph.

5. Shared Request Context

Authentication identity, headers, request IDs, application state, services, and logging are available through MCPContext.

6. Policy Enforcement

Combine default-deny authentication, scopes, confirmation, per-capability rate limits, read-only hints, and availability callbacks.

7. Backend Interface

The MCPBackend protocol isolates registration, ASGI creation, and in-process calls from the selected protocol engine.

8. Deterministic Testing

Call tools, resources, and prompts in memory while exercising the same AppMCP policy and dependency pipeline.

---

Configuration

Core application settings are supplied when creating AppMCP:

mcp = AppMCP(
    app,
    name="PaperTrail MCP",
    path="/mcp",
    auth=BearerAuth(validate_token),
    default_deny=True,
    expose_errors=False,
    stateless_http=True,
    json_response=False,
)

For a custom backend:

mcp = AppMCP(name="Store", backend=my_backend)

The backend must implement tool, resource, and prompt registration, ASGI app creation, and in-process tool calls.

---

Project Structure

appmcp/
  __init__.py              # Public API
  application.py           # AppMCP orchestration and invocation pipeline
  decorators.py            # Framework-independent decorators
  definitions.py           # Capability definitions
  registry.py              # Explicit capability registry
  context.py               # MCPContext
  dependencies.py          # Dependency container and Depends marker
  settings.py              # Runtime settings
  exceptions.py            # Public exception hierarchy
  audit.py                 # Audit events and sinks
  telemetry.py             # Optional OpenTelemetry integration
  sessions.py              # Memory and Redis session stores
  plugins.py               # Entry-point plugin loading
  cli.py                   # Inspect, dev, test, and config commands
  adapters/
    base.py                # MCPBackend protocol
    fastmcp.py             # Default FastMCP backend
    official_sdk.py        # Official SDK compatibility adapter
  integrations/
    asgi.py                # Generic ASGI mounting and lifespan composition
    fastapi.py             # FastAPI helper
    starlette.py           # Starlette helper
    flask.py               # Experimental Flask wrapper
    django.py              # Experimental Django helper
  security/
    auth.py                # Bearer and OAuth authentication callbacks
    policies.py            # Scope, confirmation, and availability policies
    rate_limits.py         # In-memory and Redis rate limiters
  testing/
    client.py              # In-memory test client
tests/
  test_*.py                # Unit and integration tests
docs/
  guide.md                 # Extended usage guide
pyproject.toml             # Package metadata and dependencies

---

Design Boundaries

• AppMCP does not implement MCP itself; the selected backend owns the protocol.
• Dynamic enabled callbacks control invocation; visible callbacks filter AppMCP capability discovery and schema output.
• The default rate limiter is process-local; coordinated deployments should use the Redis-backed limiter.
• Memory and Redis stores synchronize Mcp-Session-Id creation, sliding expiry, identity binding, confirmation state, and invalidation at the transport edge.
• Flask mounting returns an ASGI wrapper and requires an ASGI server.
• Flask, Django, and the official SDK adapter remain experimental in v1. Redis sessions and rate limits are supported for multi-worker ASGI deployments.

---

Development

# Install development and FastAPI extras
pip install -e ".[dev,fastapi]"

# Run tests
pytest -q

# Run lint checks
ruff check .

# Build distributions
python -m build

# Check package metadata
twine check dist/*

---

License

MIT

---

Contributing

Contributions are welcome. Open an issue with the host framework, transport, expected capability behavior, and a minimal example. Security reports should avoid including real tokens, credentials, or customer data.

---

Citation

If you use AppMCP in research, please cite:

@software{AppMCP2026,
  title={AppMCP: Secure Embedded MCP Servers for Python Applications},
  author={Robert McMenemy},
  url={https://github.com/Arkay92/AppMCP},
  year={2026},
  version={1.0.1},
}

---

Acknowledgments

• Model Context Protocol for the open protocol.
• FastMCP for the default MCP engine and transport.
• FastAPI and Starlette for ASGI application integration.