Build an MCP server end to end
Build a governed Model Context Protocol server with tools, authorization, observability, testing, and a production-readiness checklist.
What you will build
An AI host connects to an MCP server. The server authenticates the caller, checks policy, invokes a safe domain operation, and emits an auditable event.
AI host → MCP server → policy check → domain API
└──────────────→ audit eventPrerequisites
- Python 3.11 or newer
- Basic JSON and REST knowledge
- A terminal and code editor
- An optional MCP-compatible client
Use mock data and local credentials. Never place production secrets in this lab.
Step 1: Create the project
mkdir mcp-invoice-server
cd mcp-invoice-server
python -m venv .venv
source .venv/bin/activate
pip install mcp fastapi uvicorn pydantic python-dotenvOn Windows PowerShell, activate with .venv\Scripts\Activate.ps1.
Create app/server.py, app/policy.py, app/audit.py, .env.example, and requirements.txt.
Step 2: Define configuration safely
Create .env.example with environment, audit-log, and tenant settings. Copy it to .env locally and add .env to .gitignore.
Step 3: Add an authorization boundary
Authentication answers “who is calling?” Authorization answers “may this principal invoke this tool for this resource?” Require an explicit scope such as invoice.read before a tool can access invoice data.
def authorize(principal, required_scope):
if required_scope not in principal.scopes:
raise PermissionError("tool call denied by policy")Step 4: Add an audit trail
Record a correlation ID, principal, tool name, decision, resource, outcome, and downstream transaction ID. Minimize sensitive payloads and write events to a structured sink.
{
"event": "tool_call",
"principal": "lab-user",
"tool": "get_invoice_status",
"decision": "allow",
"correlation_id": "example-123"
}Step 5: Implement a read-only MCP tool
Expose get_invoice_status(invoice_id) through the MCP Python SDK. The tool should validate its input, authorize the principal, read from mock data, return a bounded response, and emit an audit event.
Start with read-only operations. Do not expose a write tool until identity, approval, idempotency, and rollback behavior are defined.
Step 6: Test the server
python -m compileall app
python -m pytest
cat audit.jsonlTest allowed calls, denied calls, malformed invoice IDs, missing records, timeouts, and repeated requests. A denied call must not reach the domain API.
Step 7: Connect an MCP client
Configure your MCP-compatible host with the local command and absolute working directory:
{
"mcpServers": {
"enterprise-invoice-server": {
"command": "python",
"args": ["-m", "app.server"],
"cwd": "/absolute/path/to/mcp-invoice-server"
}
}
}Verify that the client sees the expected tool description and that every invocation creates an audit event.
Step 8: Add remote transport and identity
For remote use, put the MCP endpoint behind a trusted gateway. Require TLS, validate token issuer and audience, enforce scopes, apply rate limits, propagate correlation IDs, and reject arbitrary or expired tokens.
- Authenticate the client and user/workload identity.
- Authorize the tool and resource separately.
- Validate arguments before execution.
- Fail closed for sensitive actions when policy is unavailable.
- Audit every trust boundary.
Step 9: Add operational controls
- Use distinct identities for agents and services.
- Set bounded timeouts and retries; make writes idempotent.
- Redact secrets and personal data from logs.
- Monitor latency, denials, errors, unusual volume, and prompt-to-action chains.
- Rotate credentials and review the tool registry.
Step 10: Production-readiness checklist
| Area | Done when |
|---|---|
| Identity | Every agent and service has a distinct identity. |
| Authorization | Tool, resource, scope, and policy checks are explicit. |
| Transport | TLS, audience validation, rate limits, and timeouts are enforced. |
| Audit | Prompt, identity, tool call, decision, outcome, and correlation ID are reconstructable. |
| Testing | Allow, deny, malformed input, timeout, and replay cases are tested. |
Project examples
Extend the lab with realistic scenarios.
Example 1: Privileged-access review
Architecture flow
Security analyst / AI agent
|
v
MCP client ---- TLS + correlation ID ----> MCP gateway
|
+-------------------+-------------------+
| | |
v v v
Identity provider Policy engine Audit stream
validate principal allow / deny every decision
| |
+-------------------+-------------------+
|
v
Privileged-access directory
|
v
Human reviewer approvalTrust boundary: the agent can collect evidence and create a review, but only the policy engine and an approved human reviewer can authorize a revocation.
Request sequence
- The analyst asks the agent to find privileged assignments.
- The MCP gateway authenticates the caller and creates a correlation ID.
- The server validates the tenant, principal, role, and requested scope.
- The policy engine permits read-only investigation.
- The MCP server reads the assignment and creates a pending review.
- The human reviewer approves or denies the proposed action.
- Only an approved revocation request reaches the access directory.
- The result and evidence are written to the audit stream.
Build tools to list privileged assignments and create access reviews. Analysts can investigate, but only an authorized reviewer can revoke access. Require human confirmation and make removal idempotent.
Step 1: Define the safe lab scenario
Review a mock user named alex.chen.test with a temporary Global Reader assignment in a mock tenant. No real identity provider or account is changed.
Step 2: Create mock assignment data
assignment_id: asg-1001
principal: alex.chen.test
role: Global Reader
scope: /
status: activeStep 3: Define the roles
- Analyst: may list assignments and create a review.
- Reviewer: may approve or deny a proposed revocation.
- Agent: may collect evidence but may not revoke access.
Step 4: Add the MCP tools
list_privileged_assignments(principal, scope)
create_access_review(assignment_id, reason)
revoke_assignment(assignment_id, approval_id)The final tool must require an approval ID issued by a human reviewer.
Step 5: Enforce authorization
Require access.read for listing, access.review for review creation, and access.revoke plus a valid approval ID for revocation. Never let the agent grant itself revoke permission.
Step 6: Run the investigation
Ask the MCP client to list active privileged assignments for Alex Chen and create a review for any assignment expiring within 30 days. Confirm that the response includes assignment ID, role, scope, expiry, and evidence.
Step 7: Create the review
The server creates rev-2001 with status pending. Creating a review must not change access.
Step 8: Require human approval
Sign in as the mock identity-reviewer and approve rev-2001. Record reviewer, timestamp, decision, and reason. A denied review must never call revoke.
Step 9: Revoke idempotently
Call revoke_assignment with asg-1001 and the approved review ID. If already revoked, return “already revoked” instead of applying a second change.
Step 10: Verify the audit trail
Confirm investigation, review creation, approval, and revocation are linked by one correlation ID. Verify that no secret or unnecessary personal data was recorded.
Step 11: Test failures
- Agent revokes without approval: denied.
- Analyst revokes: denied.
- Approval belongs to another assignment: denied.
- Audit sink is unavailable: revoke fails closed.
Step 12: Clean up
Delete local mock data and audit logs, deactivate the virtual environment, and confirm no real credentials or tenant identifiers were used.
Example 3: Customer-support data lookup
Expose customer-case lookup tools with tenant, region, and role restrictions. Return minimum necessary data and alert on bulk lookups, cross-tenant attempts, and repeated denials.
Example 2: Governed invoice approval
Add a high-impact approval tool. Require finance approval above a threshold, record the approver and transaction ID, and prevent an agent from approving its own request.
Lab outcome
You now have the core pattern from the MCP architecture article: a host requests a governed tool call, the server checks policy, the domain operation executes only when authorized, and the decision path is auditable.