Architecture position

Capability Extensions

Extend what an agent can do through capability packages, external evidence, persistent state, and standard protocols.

How do we add Skills, retrieval, memory, and external protocol capabilities to an agent?

Concept calibration

MCP

What it is: A standard protocol for connecting tools, resources, and prompts over local or remote transports.
What it is not: It is not a remote API, an agent, or the only way an agent can access tools.

Tool

What it is: A constrained capability an agent can invoke through explicit arguments, permissions, and result contracts.
What it is not: A tool is not the same as MCP; a local function, command, or HTTP client can also be a tool.

Harness

What it is: The runtime control environment that hosts models, tools, state, permissions, logs, budgets, human approval, and recovery.
What it is not: It can constrain risk and preserve evidence, but it cannot directly prevent model hallucinations.

MCP Servers, SaaS & Streaming Protocols

Implement tools, resources, state, Streamable HTTP, authentication, and client integration.

One thing to completeConnect different external capabilities for all three scenarios through MCP.

Before you begin, Forge already hasThe accepted solution from Module 12, Multi-Agent Collaboration & Orchestration, is this chapter’s starting point.

After this chapter, Forge canConnect Forge to external services through MCP, with a different tool in each scenario.

Smallest recovery pointValidate schemas and error contracts against a local unauthenticated MCP mock first.

Download this chapter labPython 3.12 · pytest · Pydantic · SQLite · deterministic Mock

Stage capability chain

01Define verifiable tool argument contracts that reject missing and unknown fields02Register MIME-typed resources and constrain reads to explicitly allowed URI prefixes03Declare template variables and validate missing or extra values before rendering04Assemble streamed responses in order and reject duplicates, gaps, and events after termination05Validate token expiry and scopes while constructing least-privilege authorization headers06Retry only safe requests with backoff and bind write requests to idempotency keys07Negotiate client/server capabilities and resume compatible sessions only from acknowledged cursors

Each step adds one verifiable capability; every later step begins from code and tests accepted in the previous step.

Concept calibration\n\nPut the concepts used in this module back inside their engineering boundaries before you work with the code.\n\n
\n\n### MCP mcp\n\n- What it is: A standard protocol for connecting tools, resources, and prompts over local or remote transports, with explicit schemas describing capabilities.\n- What it is not: It is not a remote API, an agent, or the only way to connect tools; identity, permission, and side effects still require governance after connection.\n- Relationship to adjacent concepts: Tools are executable capabilities, resources are readable information, and prompts are templates. MCP standardizes discovery and invocation, while the harness controls execution.\n- Where it lives in HeatStack Forge: Forge validates MCP capabilities, schemas, and identity scope, then registers permitted capabilities into the existing tool system rather than bypassing governance.\n- Typical misuse and correction: A common misuse exposes every server capability to the model. Use allowlists, least privilege, timeouts, audit, high-risk approval, and disconnect tests.\n\n### Tool tool\n\n- What it is: A constrained capability an agent invokes through explicit arguments, permissions, execution boundaries, and result contracts, such as a function, command, database query, or HTTP client.\n- What it is not: A tool is not MCP or any arbitrary script. A protocol can expose capability, while the harness still decides whether invocation is allowed and how side effects are handled.\n- Relationship to adjacent concepts: The agent loop selects tools, schemas constrain arguments, the harness enforces permission and isolation, and verification checks whether results advanced the goal.\n- Where it lives in HeatStack Forge: Forge stores tool contracts and risk levels, produces a change plan before execution, waits for approval on high-risk writes, and records results and recovery data.\n- Typical misuse and correction: A common misuse lets the model invent arguments from a tool name. Use strict schemas, deterministic validation, timeouts, idempotency keys, and refusal tests.\n\n### Harness harness\n\n- What it is: The runtime control environment that hosts models, tools, state, permissions, logs, budgets, human approval, isolation, and recovery.\n- What it is not: It can reduce risk, reject actions, and preserve evidence, but it cannot guarantee that model output is always correct or replace domain acceptance checks.\n- Relationship to adjacent concepts: The agent loop runs inside the harness; tool, memory, and MCP connections obey its policies; observability records execution and evaluation judges behavior.\n- Where it lives in HeatStack Forge: The Forge harness manages workspace, tools, approval queue, budget, checkpoints, audit logs, and rollback, separating model suggestions from real side effects.\n- Typical misuse and correction: A common misuse treats a system prompt as a security boundary. Enforce permissions, paths, network access, secrets, budgets, and rollback outside the model.\n\n

STAGE 01

Define verifiable tool argument contracts that reject missing and unknown fields

Start with a concrete problem

Why is checking only required fields not enough?

In the accepted solution from Module 12, Forge can orchestrate multiple agents to complete tasks, but it has not yet connected external tools through MCP.

This stage creates forge/mcp/schema.py, defines the ToolSchema dataclass, and implements the validate method to check arguments against the contract.

If only required fields are checked, an attacker or accidental call could pass undeclared arguments, such as shell, injecting extra capabilities.

The test test_unknown_arguments_are_rejected requires that passing a shell argument raises a ValueError containing unknown.

Make a prediction

When implementing ToolSchema.validate, what happens if you only check the required fields and ignore the allowed fields?

All valid arguments pass validation, and unknown arguments are also accepted
Unknown arguments are silently ignored
Validation raises a TypeError
Only missing required fields cause an error

Reasoning guide: The correct answer is the first option: checking only required fields lets unknown arguments through because the validation logic does not reject keys that are not in the allowed tuple.

What it is

Tool schema is a contract for argument shape

ToolSchema uses the required tuple to declare argument names that must be present, and the allowed tuple to declare all acceptable argument names.

The validate method first checks for missing required arguments, then checks for unknown arguments, and finally returns a copy of the arguments.

This contract only validates argument shape; it does not handle execution permissions or side effects.

What it is not

Tool schema is not a security boundary

Passing schema validation does not mean the tool call is safe, because execution environment, permissions, and side effects are still controlled by the Harness.

It cannot prevent the model from producing incorrect argument values; it only rejects inputs with invalid shape.

It is not the MCP protocol itself, but part of an MCP tool definition.

How it relates to neighboring concepts

Relationship between schema, Harness, and Agent Loop

After the Agent Loop selects a tool, the Harness uses ToolSchema.validate to check arguments before deciding whether to allow the call.

The schema’s allowed tuple limits the set of acceptable arguments, preventing unintended capability injection.

After validation passes, the Harness can still reject execution based on policy, for example if dangerous argument values are detected.

Boundary decision for this stage: Tool schema only validates argument shape, not execution permissions or side effects; therefore, even if validation passes, the Harness must independently check permissions and side effects.

Experiment: mcp-saas-streaming-01-tool-schema-contract
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/mcp-saas-streaming/01-tool-schema-contract/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how ToolSchema rejects unknown arguments: construct a schema allowing query and limit, pass arguments including shell, observe ValueError and explain why.

Suppose we define a tool search_docs that accepts only query and limit arguments.

If a caller passes a shell argument, the validator must reject it because shell is not in the allowed tuple.

In the implementation, we use a list comprehension to find all keys not in allowed, and if the list is non-empty, raise a ValueError.

Implementation steps

Create a ToolSchema instance with required=("query",) and allowed=("query", "limit").
Call validate with {"query": "release policy", "shell": "rm -rf"}.
The validator finds shell is not in allowed, raises ValueError with a message containing unknown.

Verified solution code: forge/mcp/__init__.py

"""MCP protocol building blocks."""

Verified solution code: forge/mcp/schema.py

from dataclasses import dataclass

@dataclass(frozen=True)
class ToolSchema:
    name: str
    required: tuple[str, ...]
    allowed: tuple[str, ...]

    def validate(self, arguments: dict[str, object]) -> dict[str, object]:
        missing = [key for key in self.required if key not in arguments]
        unknown = [key for key in arguments if key not in self.allowed]
        if missing:
            raise ValueError("missing required arguments: " + ", ".join(missing))
        if unknown:
            raise ValueError("unknown arguments: " + ", ".join(unknown))
        return dict(arguments)

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Why this implementation

The validate method first checks for missing required arguments; here query is present, so there is no missing argument.

Then it checks for unknown arguments: shell is not in the allowed tuple, so it is added to the unknown list.

Since the unknown list is non-empty, a ValueError is raised with a message containing unknown and the argument name.

Guided practice

Complete the missing unknown-argument check in ToolSchema.validate to make test test_unknown_arguments_are_rejected pass.

Open forge/mcp/schema.py and locate the validate method.
After checking for missing arguments, add code to find all argument keys not in self.allowed.
If the unknown argument list is non-empty, raise a ValueError with message format unknown arguments: followed by comma-separated argument names.
Run python -m pytest -q tests/test_stage.py to verify both tests pass.

Progressive hints

Use a list comprehension: [key for key in arguments if key not in self.allowed].
The error message should contain the word unknown to satisfy the test assertion.
Make sure to check for unknown arguments before returning the argument copy.

Success criteria

test_valid_arguments_are_copied passes, verifying that valid arguments return a copy.
test_unknown_arguments_are_rejected passes, verifying that unknown arguments raise a ValueError.

Fault injection and diagnosis

The current validate method in forge/mcp/schema.py only checks for missing required arguments and does not check for unknown arguments. Therefore, when the test test_unknown_arguments_are_rejected passes a shell argument, the validator incorrectly accepts it, causing the test to fail with AssertionError: unknown tool argument was accepted.

This stage has taught the role of the allowed tuple in ToolSchema and how to find unknown arguments using a list comprehension. Therefore, we can locate the missing unknown-argument check in the validate method and add the appropriate code to fix the defect.

Verified fault contract

Target test: test_unknown_arguments_are_rejected
Faulty file: forge/mcp/schema.py

Diagnosis order

Run python -m pytest -q ../tests/test_stage.py::test_unknown_arguments_are_rejected and observe the test failure.
Inspect the validate method in forge/mcp/schema.py and notice that it only checks for missing arguments, not unknown arguments.
Confirm that the allowed tuple contains query and limit, but the passed shell is not in it.
Conclude that an unknown-argument check is needed to reject keys not in allowed.

Fix principle: Add an unknown-argument check in the validate method: compute unknown = [key for key in arguments if key not in self.allowed], and if unknown is non-empty, raise ValueError("unknown arguments: " + ", ".join(unknown)).

python -m pytest -q ../tests/test_stage.py::test_unknown_arguments_are_rejected

Transfer challenge

In the office research scenario, you need to use ToolSchema to define the tool search_docs with required argument query and allowed arguments limit and sort, and verify that passing an extra argument file_type raises ValueError.

Changed condition: The allowed argument set is expanded from (‘query’, ‘limit’) to (‘query’, ‘limit’, ‘sort’), and an extra argument ‘file_type’ is passed to verify rejection behavior.

Write a Python script that creates a ToolSchema instance with name ‘search_docs’, required=(‘query’,), allowed=(‘query’, ‘limit’, ‘sort’). Then call validate with arguments {‘query’: ‘quarterly report’, ‘sort’: ‘date’} and verify it returns a copy; then call validate with arguments {‘query’: ‘quarterly report’, ‘file_type’: ‘pdf’}, catch the ValueError exception and print the exception message.

Success criteria

The script successfully creates a ToolSchema instance with allowed containing ‘sort’.
Calling validate with arguments containing ‘sort’ returns a copy of the arguments without raising an exception.
Calling validate with arguments containing ‘file_type’ catches a ValueError exception, and the exception message contains ‘unknown’ and ‘file_type’.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Design explanation question: Why is checking only required fields insufficient to prevent unknown arguments? Explain with reference to the allowed tuple in ToolSchema.

A strong answer should include

Checking required fields only ensures that arguments in required are present, but it does not prevent callers from passing undeclared arguments.
The allowed tuple defines all acceptable argument names, and the validator must reject keys not in it, otherwise unintended capabilities may be introduced.
Unknown arguments could be misused by downstream code, leading to security vulnerabilities or unexpected behavior.

Mastery evidence

ToolSchema.validate returns a copy of valid arguments, and test test_valid_arguments_are_copied passes.
ToolSchema.validate raises a ValueError for unknown arguments, and test test_unknown_arguments_are_rejected passes.
In the office research scenario, ToolSchema is used to define search_docs and verify rejection of the extra argument file_type.

STAGE 02

Register MIME-typed resources and constrain reads to explicitly allowed URI prefixes

Start with a concrete problem

How can we prevent resource registration from escaping allowed URI prefixes?

Currently forge/mcp/schema.py exists, but there is no resource registry. The test file tests/test_stage.py imports forge.mcp.resources, but that module has not been created, causing a collection error: ModuleNotFoundError: No module named 'forge.mcp.resources'.

You need to create forge/mcp/resources.py and implement the Resource dataclass and ResourceRegistry class. ResourceRegistry receives a tuple of allowed URI prefixes in its constructor and must enforce that registered resources have URIs starting with one of these prefixes.

If a registered URI is not within the allowed prefixes, a ValueError must be raised with the word “allowed” in the message. The registry must also support reading registered resources and preserve their MIME type.

Make a prediction

In ResourceRegistry.register, if the passed resource.uri does not start with any prefix in self.allowed_prefixes, what should happen?

Store the resource anyway, because the registry only stores and does not validate.
Raise ValueError indicating the URI is outside allowed prefixes.
Silently ignore the resource without storing or raising an error.
Automatically convert the URI to a path under an allowed prefix before storing.

Reasoning guide: The correct option is to raise ValueError. The registry must enforce the prefix policy; otherwise out-of-scope resources would be accepted, potentially exposing content that should not be accessible. Silently ignoring or auto-converting would mask errors and prevent callers from detecting configuration problems.

What it is

What ResourceRegistry is

ResourceRegistry is an in-memory container for resources that maintains a mapping from URI to Resource objects and enforces URI prefix policy at registration time.

It defines an acceptable URI namespace through the allowed_prefixes tuple, for example ("forge://workspace/",) means only URIs starting with forge://workspace/ are allowed.

The register method checks whether the resource URI starts with any allowed prefix; if not, it raises ValueError, preventing out-of-scope resources from entering the registry.

What it is not

What ResourceRegistry is not

It is not a content parser or transport layer: it does not care about the specific format of resource content and does not send resources to clients.

It is not a security boundary by itself: it only performs prefix checks, but callers must still ensure allowed_prefixes is configured correctly and that reads also validate the URI.

It is not a global resource manager: each registry instance maintains its own mapping independently, and data is not shared between instances.

How it relates to neighboring concepts

Relationships with other components

The Resource dataclass is an immutable object with three fields: uri, mime_type, and content; the registry stores references to these objects.

ResourceRegistry depends on allowed_prefixes to define its boundary, and both register and read methods use this boundary for validation.

The test test_registered_resource_preserves_media_type verifies that a registered resource’s MIME type is preserved after reading, while test_resource_outside_allowed_prefix_is_rejected verifies that out-of-scope resources are rejected.

Boundary decision for this stage: The registry only handles storage and prefix policy, not content parsing or transport. Therefore, register must check URI prefixes to prevent out-of-scope resources from being accepted; read must also check URI prefixes to prevent callers from directly reading unregistered out-of-scope URIs.

Experiment: mcp-saas-streaming-02-resource-registry
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/mcp-saas-streaming/02-resource-registry/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how ResourceRegistry rejects out-of-scope resources: create a registry with allowed prefix forge://workspace/, attempt to register file:///etc/passwd, observe ValueError.

Suppose you are building an MCP server that needs to expose resources within a workspace. You create a ResourceRegistry that only allows the forge://workspace/ prefix.

Now someone tries to register a local file file:///etc/passwd, which is not within the allowed prefix. If the registry does not check the prefix, this sensitive file would be exposed.

The correct implementation should check in the register method whether the URI starts with any prefix in self.allowed_prefixes, and if not, raise ValueError.

Implementation steps

Create a ResourceRegistry instance with ("forge://workspace/",) as allowed prefixes.
Construct a Resource object with URI file:///etc/passwd, MIME type text/plain, and content b"secret".
Call registry.register(resource) and expect a ValueError because the URI is not within allowed prefixes.

Verified solution code: forge/mcp/resources.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Resource:
    uri: str
    mime_type: str
    content: bytes

class ResourceRegistry:
    def __init__(self, allowed_prefixes: tuple[str, ...]):
        self.allowed_prefixes = allowed_prefixes
        self._items: dict[str, Resource] = {}

    def register(self, resource: Resource) -> None:
        if not resource.uri.startswith(self.allowed_prefixes):
            raise ValueError("resource URI is outside allowed prefixes")
        if resource.uri in self._items:
            raise ValueError("resource URI must be unique")
        self._items[resource.uri] = resource

    def read(self, uri: str) -> Resource:
        if not uri.startswith(self.allowed_prefixes):
            raise PermissionError("resource URI is outside allowed prefixes")
        return self._items[uri]

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Why this implementation

The test expects a ValueError when registering an out-of-scope resource, but the current implementation does not check the prefix and stores the resource directly, causing the test to fail.

The fix is to add a prefix check at the beginning of the register method: if not resource.uri.startswith(self.allowed_prefixes): raise ValueError("resource URI is outside allowed prefixes").

This way, any URI not within the allowed prefixes is rejected, ensuring the registry only contains resources from authorized namespaces.

Guided practice

Add URI prefix check in ResourceRegistry.register to make test test_resource_outside_allowed_prefix_is_rejected pass.

Open forge/mcp/resources.py (create it if it does not exist).
Define the Resource dataclass with fields uri, mime_type, content and decorate with @dataclass(frozen=True).
Define the ResourceRegistry class with constructor accepting allowed_prefixes: tuple[str, ...] and storing it, and initialize an empty dictionary self._items.
Implement the register method: first check if resource.uri starts with any prefix in self.allowed_prefixes; if not, raise ValueError with a message containing “allowed”.
Implement the read method: check if uri starts with an allowed prefix; if not, raise PermissionError, then return the corresponding resource from self._items.

Progressive hints

Use str.startswith(tuple) to check multiple prefixes at once.
The error message must contain the word “allowed”, otherwise the test assertion will fail.
In read, if the URI is not within allowed prefixes, raise PermissionError instead of ValueError to distinguish between registration and read errors.

Success criteria

Running python -m pytest -q tests/test_stage.py results in both tests passing.
test_resource_outside_allowed_prefix_is_rejected no longer raises AssertionError; instead it successfully catches ValueError.

Fault injection and diagnosis

The failure fixture fixtures/failure.json describes a resource scope bypass defect: the registry accepts resources outside the configured URI namespace. In forge/mcp/resources.py, the register method lacks prefix enforcement, causing a file:// resource to be accepted by a forge:// registry.

The core learning outcome of this stage is to implement URI prefix constraints. The defect in the failure fixture is exactly the missing constraint, and you have already learned through guided practice how to add prefix checks in register. Therefore, you can directly locate the register method, add the missing validation logic, and make the test test_resource_outside_allowed_prefix_is_rejected pass.

Verified fault contract

Target test: test_resource_outside_allowed_prefix_is_rejected
Faulty file: forge/mcp/resources.py

Diagnosis order

Run the faulty command python -m pytest -q ../tests/test_stage.py::test_resource_outside_allowed_prefix_is_rejected and observe the test failure: AssertionError: out-of-scope resource was registered.
Inspect the register method in forge/mcp/resources.py and notice it directly executes self._items[resource.uri] = resource without checking the URI prefix.
Compare with the test expectation: the test attempts to register file:///etc/passwd and expects a ValueError, but none is raised, indicating the prefix check is missing.
Fix: add at the beginning of register: if not resource.uri.startswith(self.allowed_prefixes): raise ValueError("resource URI is outside allowed prefixes").

Fix principle: In the ResourceRegistry.register method, before storing the resource, check whether resource.uri starts with any prefix in self.allowed_prefixes. If not, raise ValueError with a message containing “allowed”. This way, out-of-scope resources are rejected at registration, and the test test_resource_outside_allowed_prefix_is_rejected will pass.

python -m pytest -q ../tests/test_stage.py::test_resource_outside_allowed_prefix_is_rejected

Transfer challenge

In the music creation scenario, you need to use ResourceRegistry to manage lyric files. The allowed prefix is forge://workspace/, and you register a resource forge://workspace/lyrics.txt with MIME type text/plain and content b"La la la". Then you attempt to register an external URL https://example.com/lyrics.txt, expecting a ValueError.

Changed condition: Change the resource URI from a workspace path to an external HTTPS URL to test whether the prefix constraint still holds.

Write a Python script or interactive session that creates a ResourceRegistry instance with allowed prefixes ("forge://workspace/",). Register Resource("forge://workspace/lyrics.txt", "text/plain", b"La la la"), then attempt to register Resource("https://example.com/lyrics.txt", "text/plain", b"stolen") and catch the exception.

Success criteria

Registering forge://workspace/lyrics.txt succeeds, and a subsequent read returns MIME type text/plain.
Attempting to register https://example.com/lyrics.txt raises ValueError with a message containing “allowed”.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Design explanation question: Why check URI prefixes both at registration and read time? If you only check in register but not in read, what security risk arises?

A strong answer should include

Checking in register prevents out-of-scope resources from entering the registry, ensuring stored data is within authorized namespaces.
Checking in read prevents callers from directly requesting unregistered out-of-scope URIs, even if the registry does not contain that URI, thereby avoiding information leakage or erroneous access.
If only register checks, an attacker could construct a URI not in the registry and call read directly; if read does not check the prefix, it might return unexpected results or expose internal implementation details.

Mastery evidence

ResourceRegistry.register rejects URIs outside allowed prefixes; test test_resource_outside_allowed_prefix_is_rejected passes.
ResourceRegistry.read returns the registered resource with correct MIME type; test test_registered_resource_preserves_media_type passes.
Running python -m pytest -q tests/test_stage.py results in both tests passing with exit code 0.

STAGE 03

Declare template variables and validate missing or extra values before rendering

Start with a concrete problem

Missing variables are silently replaced with empty strings, producing incomplete prompts

In the previous stage, we implemented ResourceRegistry, but there is no prompt template yet. Now we need to create forge/mcp/prompts.py and implement the PromptTemplate class.

The test test_missing_prompt_variable_is_rejected requires that when rendering a prompt template, if a declared variable is missing, a ValueError must be raised instead of silently filling it with an empty string.

Currently, forge/mcp/prompts.py does not exist in the starting code, so running the tests will raise ModuleNotFoundError. We need to implement PromptTemplate from scratch and ensure the variable validation logic is correct.

Make a prediction

What happens if the render method of PromptTemplate fills missing variables with empty strings?

The rendered result is incomplete, but no error is raised, and the test fails
The rendered result is complete, and the test passes
A ValueError is raised, and the test passes
The rendered result is incomplete, but the test passes

Reasoning guide: The correct answer is the first option. If missing variables are filled with empty strings, the rendered result will lack necessary information, and the test test_missing_prompt_variable_is_rejected will fail because it expects a ValueError to be raised.

What it is

PromptTemplate is a variable contract and renderer

PromptTemplate declares the set of variables required for rendering and validates that the provided values exactly match before rendering.

It explicitly lists all required variables through the variables tuple, and the render method checks for missing and extra variables to ensure the contract is honored.

Rendering uses Python’s str.format mechanism to replace placeholders in the template with the provided values.

What it is not

PromptTemplate does not handle prompt content quality or model invocation

PromptTemplate does not evaluate whether the prompt text is effective or high-quality; it only guarantees the completeness of variable substitution.

It does not involve model calls, API requests, or any external side effects; it is purely a data transformation component.

It does not handle type conversion or default values for variables; all variables must be explicitly provided.

How it relates to neighboring concepts

Relationship between PromptTemplate and ResourceRegistry

ResourceRegistry manages readable resources, while PromptTemplate manages renderable prompt templates; both are part of MCP capabilities.

PromptTemplate can reference resource content from ResourceRegistry as variable values, but variable validation is independent of resource existence.

Both follow explicit schema constraints, but PromptTemplate focuses on the variable contract, while ResourceRegistry focuses on resource discovery.

Boundary decision for this stage: PromptTemplate only handles the variable contract and rendering, not prompt content quality or model invocation; missing or extra variables must raise ValueError instead of being silently handled.

Experiment: mcp-saas-streaming-03-prompt-template-contract
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/mcp-saas-streaming/03-prompt-template-contract/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how PromptTemplate rejects missing variables: create template “Summarize {topic} for {audience}.”, provide only topic, observe ValueError.

Suppose we have a prompt template “Summarize {topic} for {audience}.” that declares variables topic and audience.

If we provide only the value for topic and omit audience, rendering should raise a ValueError instead of producing an incomplete prompt like “Summarize MCP for .”.

In the implementation, we first compute the missing variables set missing = set(self.variables) - set(values), then check if it is empty.

If missing is non-empty, raise a ValueError with a message containing the keyword “missing” to satisfy the test assertion.

Implementation steps

Create a PromptTemplate instance: prompt = PromptTemplate(“brief”, “Summarize {topic} for {audience}.”, (“topic”, “audience”))
Call prompt.render({“topic”: “MCP”}), providing only topic and omitting audience.
In the render method, compute missing = {“audience”} and detect the missing variable.
Raise a ValueError with a message like “prompt variables mismatch missing=[‘audience’] extra=[]”.

Verified solution code: forge/mcp/prompts.py

from dataclasses import dataclass

@dataclass(frozen=True)
class PromptTemplate:
    name: str
    template: str
    variables: tuple[str, ...]

    def render(self, values: dict[str, str]) -> str:
        missing = set(self.variables) - set(values)
        extra = set(values) - set(self.variables)
        if missing or extra:
            raise ValueError(f"prompt variables mismatch missing={sorted(missing)} extra={sorted(extra)}")
        return self.template.format(**values)

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Why this implementation

The missing variable check ensures that all declared variables are provided, preventing incomplete prompts.

Using set difference to compute missing variables is simple and efficient, and clearly reports which variables are missing.

The error message contains the keyword ‘missing’ to satisfy the test’s assertion.

Guided practice

Add missing and extra variable checks in PromptTemplate.render to make test_missing_prompt_variable_is_rejected pass.

Create forge/mcp/prompts.py and define the PromptTemplate dataclass with fields name, template, and variables.
Implement the render method: first compute the missing variables set missing = set(self.variables) - set(values).
Compute the extra variables set extra = set(values) - set(self.variables).
If missing or extra is non-empty, raise a ValueError with a message containing ‘missing’ and ‘extra’ information.
Otherwise, render the template using self.template.format(**values) and return the result.

Progressive hints

Use set operations to easily find missing and extra keys.
The error message must contain the string ‘missing’ because the test asserts it.
Use the format method instead of format_map so that a KeyError is raised if variables do not match.

Success criteria

When running python -m pytest -q tests/test_stage.py, test_missing_prompt_variable_is_rejected passes.
test_declared_variables_render_deterministically also passes, ensuring normal rendering is unaffected.

Fault injection and diagnosis

The fault fixture replaces strict validation in render with empty-string defaults. When audience is missing, the template still returns an incomplete prompt, so test_missing_prompt_variable_is_rejected fails.

This stage established the variable contract: the variables declared by a template must exactly match the values supplied by the caller. The fault breaks that contract, so comparing the two sets points directly to the render boundary.

Verified fault contract

Target test: test_missing_prompt_variable_is_rejected
Faulty file: forge/mcp/prompts.py

Diagnosis order

Run python -m pytest -q ../tests/test_stage.py::test_missing_prompt_variable_is_rejected and keep the failing assertion.
Inspect forge/mcp/prompts.py and confirm that render supplies an empty string for a missing key instead of rejecting the input.
Compare set(self.variables) with set(values); the missing audience key explains why an incomplete prompt is returned.
Restore strict missing and extra checks, then render only when both sets are empty.

Fix principle: In render, compute missing = set(self.variables) - set(values) and extra = set(values) - set(self.variables). Raise ValueError when either set is non-empty; otherwise call self.template.format(**values).

python -m pytest -q ../tests/test_stage.py::test_missing_prompt_variable_is_rejected

Transfer challenge

Office research: render a deadline-aware briefing prompt and verify both sides of the variable contract.

Changed condition: The scenario changes from software delivery to office research, and the template contract expands from topic/audience to topic/audience/deadline; the PromptTemplate API and validation rule stay unchanged.

Using the existing PromptTemplate API, create the template “Prepare a brief on {topic} for {audience} by {deadline}.” with declared variables topic, audience, and deadline. First render it with all three values. Then omit deadline and verify ValueError contains missing; finally add undeclared location and verify ValueError contains extra.

Success criteria

The complete input renders a deterministic briefing prompt containing the topic, audience, and deadline.
Omitting deadline raises ValueError and the message contains missing.
Adding undeclared location raises ValueError and the message contains extra; use the existing PromptTemplate API and tests.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Design explanation question: Why must extra variables also be rejected? Explain the significance of rejecting extra variables for the integrity of the prompt template contract.

A strong answer should include

Extra variables may indicate that the caller misunderstands the template’s variable contract or is passing irrelevant data.
Rejecting extra variables prevents accidentally ignoring important information and ensures the rendered result meets expectations.
It forces callers to precisely match the declared variables, improving code robustness and maintainability.

Mastery evidence

PromptTemplate.render returns a deterministic string for valid variables, such as ‘Summarize MCP for engineers.’.
PromptTemplate.render raises a ValueError for missing variables, with an error message containing ‘missing’.
PromptTemplate.render raises a ValueError for extra variables, with an error message containing ‘extra’.

STAGE 04

Assemble streamed responses in order and reject duplicates, gaps, and events after termination

Start with a concrete problem

Stream events must arrive in order and stop after termination

In streaming protocols, events may arrive out of order due to network retransmission, concurrent pushes, or client errors, but the final concatenated result must match the original sending order.

If the assembler blindly accepts any sequence number, it will incorrectly concatenate data after a missing event, producing unverifiable output.

This stage requires implementing StreamAssembler, which only accepts contiguous increasing sequence numbers and rejects any events after receiving a done event.

In the current starting code, the file forge/mcp/streaming.py does not exist, so the test test_sequence_gap_is_rejected fails with an import error.

Make a prediction

If StreamAssembler.push receives sequence number 3 while the expected sequence number is 2, what should happen?

Accept the event and update the expected sequence to 4
Raise ValueError because the sequence is not contiguous
Ignore the event and wait for sequence 2
Automatically insert an empty event and then accept

Reasoning guide: The correct behavior is to raise ValueError, because streaming protocols require events to arrive strictly in order; any gap indicates data loss or reordering and must be explicitly rejected rather than silently repaired.

What it is

What StreamAssembler is

StreamAssembler is a stateful object that maintains next_sequence (the next expected sequence number), chunks (a list of received text chunks), and finished (whether a termination event has been received).

Its push method validates event sequence continuity, event type legality, and termination state before applying the event data.

The text method returns the concatenated result only when finished is true; otherwise it raises an exception to prevent returning incomplete content.

What it is not

What StreamAssembler is not

It does not handle network transport, retransmission, or buffering of out-of-order events; those belong to the transport layer or a higher-level protocol.

It does not parse event content or validate business semantics; it only focuses on sequence numbers and termination state.

It is not thread-safe; if multiple threads push events concurrently, external synchronization is required.

How it relates to neighboring concepts

Relationships with other components

StreamAssembler uses next_sequence as a cursor, comparing it with the sequence number in push to ensure that only the expected next event is accepted.

The finished state is set by the done event; once true, any subsequent push immediately raises an exception.

The chunks list stores text blocks in the order received, and the text method concatenates them into the final string after termination.

Boundary decision for this stage: StreamAssembler only handles ordering and termination state, not network transport or retransmission; any out-of-order or post-termination events must be rejected rather than repaired.

Experiment: mcp-saas-streaming-04-ordered-stream-events
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/mcp-saas-streaming/04-ordered-stream-events/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how StreamAssembler rejects sequence gaps: push sequence 1 then sequence 3, observe ValueError.

We start with an empty StreamAssembler whose next_sequence is initially 1.

First, we push a chunk event with sequence 1 and data “hello”; this updates next_sequence to 2.

Then we attempt to push a done event with sequence 3; since the expected sequence is 2, the push method detects the discontinuity and raises ValueError.

Implementation steps

Create a StreamAssembler instance.
Call push(1, "chunk", "hello"), which successfully adds the data and updates next_sequence to 2.
Call push(3, "done"), which triggers the sequence continuity check and raises ValueError.

Verified solution code: forge/mcp/streaming.py

from dataclasses import dataclass, field

@dataclass
class StreamAssembler:
    next_sequence: int = 1
    chunks: list[str] = field(default_factory=list)
    finished: bool = False

    def push(self, sequence: int, kind: str, data: str = "") -> None:
        if self.finished:
            raise ValueError("stream is already terminal")
        if sequence != self.next_sequence:
            raise ValueError("stream sequence must be contiguous")
        if kind not in {"chunk", "done"}:
            raise ValueError("unknown stream event")
        if kind == "chunk":
            self.chunks.append(data)
        else:
            self.finished = True
        self.next_sequence += 1

    def text(self) -> str:
        if not self.finished:
            raise ValueError("stream is incomplete")
        return "".join(self.chunks)

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Why this implementation

The push method first checks the finished state, then compares the incoming sequence with self.next_sequence.

If they are not equal, it immediately raises ValueError, preventing the event from being applied and protecting the internal state from corruption.

Guided practice

Add sequence continuity check in StreamAssembler.push to make test_sequence_gap_is_rejected pass.

Create the file forge/mcp/streaming.py and define the StreamAssembler dataclass.
In the push method, after checking finished, add a sequence continuity check: if sequence != self.next_sequence, raise ValueError.
Add an event type check: if kind is not chunk or done, raise ValueError.
Update state based on kind: chunk appends data, done sets finished to True.
Update next_sequence to sequence + 1.

Progressive hints

The sequence check must occur before any state modification.
The error message should contain ‘contiguous’ to pass the test assertion.
Don’t forget to handle unknown event types.

Success criteria

test_sequence_gap_is_rejected test passes.
test_stream_assembles_only_after_done test passes.

Fault injection and diagnosis

The current forge/mcp/streaming.py has a StreamAssembler.push method that lacks the sequence continuity check, causing the test test_sequence_gap_is_rejected to fail. The test expects that pushing sequence 1 then sequence 3 raises a ValueError containing ‘contiguous’, but the actual code accepts sequence 3 and updates the cursor without raising an exception.

The core mechanism of this stage is the sequence continuity check, and the learner has already understood how to compare sequence with next_sequence through the worked example and guided practice. Now they need to diagnose why the actual code is missing this check and apply the same repair pattern.

Verified fault contract

Target test: test_sequence_gap_is_rejected
Faulty file: forge/mcp/streaming.py

Diagnosis order

Run the test python -m pytest -q ../tests/test_stage.py::test_sequence_gap_is_rejected and observe the failure output.
Inspect the push method in forge/mcp/streaming.py and notice that there is no code comparing sequence with self.next_sequence.
Confirm the defect: the code directly accepts any sequence number and updates next_sequence, thus ignoring gaps.

Fix principle: In the push method, after checking finished, add if sequence != self.next_sequence: raise ValueError("stream sequence must be contiguous") to ensure only contiguous sequences are accepted.

python -m pytest -q ../tests/test_stage.py::test_sequence_gap_is_rejected

Transfer challenge

In the office research scenario, you need to assemble a stream of text chunks that form a brief. The stream will contain three ‘chunk’ events with data ’Background: ’, ‘Office Research’, and ’ Conclusion’, followed by a ‘done’ event. You must use the existing StreamAssembler to push these events in order, verify that text() returns the concatenated result, and confirm that pushing an extra chunk after done raises an exception.

Changed condition: The scenario changes from a single stream with two chunks to a stream with three chunks and a different content, while preserving the same event types and API. The learner must apply the same contiguity and termination rules to a slightly more complex sequence, reinforcing the understanding that the assembler’s behavior is independent of the number of chunks or their content.

Write a Python script that creates a StreamAssembler instance. Push three ‘chunk’ events with sequences 1, 2, and 3, and data ’Background: ’, ‘Office Research’, and ’ Conclusion’ respectively. Then push a ‘done’ event with sequence 4. After assembly, call text() and verify it returns ‘Background: Office Research Conclusion’. Finally, attempt to push a chunk event with sequence 5 after done, catch and print the ValueError message. The script should demonstrate that the assembler correctly enforces contiguity and termination.

Success criteria

The script successfully pushes the three chunks and done in order, and text() returns ‘Background: Office Research Conclusion’.
Pushing an extra chunk after done raises ValueError, and the message indicates the stream is already terminal (e.g., contains ‘terminal’).

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Design explanation question: Why must events after termination be rejected? What specific impact would allowing data to be pushed after termination have on the integrity and verifiability of the streaming response?

A strong answer should include

The termination event indicates that the stream is complete; any subsequent event implies a protocol violation or data corruption.
Allowing post-termination pushes would break the semantics of the finished state, causing text() to potentially return incomplete or incorrectly concatenated content.
Rejecting post-termination events exposes client or transport errors early, avoiding silently producing untrustworthy results.

Mastery evidence

StreamAssembler.text() returns the concatenated result only after receiving done; otherwise it raises ValueError.
StreamAssembler.push raises ValueError containing ‘contiguous’ for non-contiguous sequences.
StreamAssembler.push rejects any event after finished is true, raising ValueError containing ‘terminal’.

STAGE 05

Validate token expiry and scopes while constructing least-privilege authorization headers

Start with a concrete problem

A read-only token is misused for a write operation

In SaaS integrations, we often receive an access token (AccessToken) that carries an expiry time and a set of scopes.

The current code in AccessToken.headers_for only checks whether the token is expired, but does not check whether the requested operation is within the token’s allowed scopes.

As a result, a token with only issues:read permission can generate an Authorization header for an issues:write operation, leading to unauthorized write calls.

We need to validate both token freshness and required scope before generating the authorization header, otherwise raise PermissionError.

Make a prediction

In AccessToken.headers_for, if the token is not expired but lacks the required scope, what does the current code do?

Raises PermissionError because scope check already exists
Returns Authorization header because only expiry is checked
Returns empty dictionary because scope is missing
Raises ValueError because scope format is invalid

Reasoning guide: The current code only checks now >= self.expires_at and does not check whether required_scope is in self.scopes. Therefore, as long as the token is not expired, it returns the Authorization header even if the scope does not match.

What it is

AccessToken is a minimal authorization decision unit

AccessToken is an immutable data class that encapsulates the token value, scope set, and expiry time.

Its headers_for method performs two checks before generating the authorization header: first expiry, then scope.

This order ensures we do not waste scope checks on an already invalid token and avoids leaking the token value.

What it is not

AccessToken does not handle token acquisition or refresh

AccessToken only handles validation and header generation; it does not fetch new tokens from an OAuth server or refresh expired tokens.

It also does not perform actual HTTP calls or manage global permission policies; it only makes a local decision for a single request’s required scope.

How it relates to neighboring concepts

Order relationship between expiry and scope checks

Expiry check is a precondition: if the token is expired, it cannot be used regardless of scope, so we raise an expiry error first.

Scope check is authorization: only when the token is unexpired and contains the required scope do we generate the Authorization header.

sanitized_headers is independent of AccessToken; it is used to redact the authorization header in logs to prevent token value leakage.

Boundary decision for this stage: Before generating the Authorization header, both conditions must be satisfied: the token is unexpired and contains the required scope; if either fails, raise PermissionError instead of returning a partial or empty header.

Experiment: mcp-saas-streaming-05-scoped-saas-auth
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/mcp-saas-streaming/05-scoped-saas-auth/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how AccessToken rejects missing scope: create a read-only token, attempt to generate headers for issues:write, observe PermissionError.

Suppose we have a read-only token with scopes {"issues:read"} and expiry time 200.

At current time 100, the token is not expired. We attempt to generate headers for the issues:write operation.

Since issues:write is not in the token’s scope set, headers_for should raise PermissionError instead of returning headers.

Implementation steps

Create AccessToken("secret", frozenset({"issues:read"}), 200).
Call token.headers_for("issues:write", 100).
Observe that PermissionError is raised and the error message contains “scope”.

Verified solution code: forge/mcp/auth.py

from dataclasses import dataclass

@dataclass(frozen=True)
class AccessToken:
    value: str
    scopes: frozenset[str]
    expires_at: int

    def headers_for(self, required_scope: str, now: int) -> dict[str, str]:
        if now >= self.expires_at:
            raise PermissionError("access token is expired")
        if required_scope not in self.scopes:
            raise PermissionError("required scope is missing")
        return {"Authorization": f"Bearer {self.value}"}

def sanitized_headers(headers: dict[str, str]) -> dict[str, str]:
    return {key: ("<redacted>" if key.lower() == "authorization" else value) for key, value in headers.items()}

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Why this implementation

In headers_for, first check now >= self.expires_at; here 100 < 200, so no expiry error.

Then check required_scope not in self.scopes; since issues:write is not in {"issues:read"}, it raises PermissionError.

This check prevents a read-only token from being used for write operations, implementing the principle of least privilege.

Guided practice

Add scope check in AccessToken.headers_for to make test_missing_scope_is_rejected pass.

Open the file forge/mcp/auth.py.
In the headers_for method, after the expiry check, add a scope check: if required_scope not in self.scopes, raise PermissionError("required scope is missing").
Run the test python -m pytest -q tests/test_stage.py::test_missing_scope_is_rejected and confirm it passes.

Progressive hints

The scope check should come after the expiry check because an expired token cannot be used regardless of scope.
Use the not in operator to check whether required_scope is in self.scopes.
The error message should contain “scope” to satisfy the test assertion.

Success criteria

The test test_missing_scope_is_rejected passes.
headers_for raises PermissionError when the scope is missing, and the message contains “scope”.

Fault injection and diagnosis

The fault fixture fixtures/failure.json injects a defect: AccessToken.headers_for lacks the scope check. When running the test test_missing_scope_is_rejected, the test fails because a read-only token successfully generates an authorization header for a write operation. We need to locate and fix this defect.

This stage has just taught how to add a scope check in headers_for. The defect in the fault fixture is exactly the missing check, so we can directly apply the learned knowledge to fix it. By observing the test failure, we can confirm the defect location and add the correct check.

Verified fault contract

Target test: test_missing_scope_is_rejected
Faulty file: forge/mcp/auth.py

Diagnosis order

Run python -m pytest -q ../tests/test_stage.py::test_missing_scope_is_rejected and observe the test failure.
Look at the test code: it creates a read-only token, then attempts to generate headers for issues:write, expecting a PermissionError.
Inspect the headers_for method in forge/mcp/auth.py and find that only the expiry check is present, with no scope check.
Confirm the defect: missing if required_scope not in self.scopes: raise PermissionError(...).

Fix principle: In the headers_for method, after the expiry check, add a scope check: if required_scope not in self.scopes: raise PermissionError("required scope is missing"). This way, when the required scope is not in the token’s scopes, a PermissionError is raised and the test passes.

python -m pytest -q ../tests/test_stage.py::test_missing_scope_is_rejected

Transfer challenge

In the music creation scenario, you have a read-only token with scopes {"music:read"} and expiry time 300. The current time is 150. You need to verify that generating headers for music:write raises PermissionError, and check that sanitized_headers redacts the Authorization header.

Changed condition: Change the scope from issues:read to music:read, and the required operation from issues:write to music:write, while adjusting expiry and current times.

Create an AccessToken instance with value "music-secret", scopes frozenset({"music:read"}), and expiry time 300. Then call headers_for("music:write", 150) and confirm that it raises PermissionError. Next, use a valid read-only token (e.g., scopes {"music:read"}) to generate headers for music:read, and call sanitized_headers to check that the Authorization value is replaced with <redacted>.

Success criteria

Calling headers_for("music:write", 150) raises PermissionError and the error message contains “scope”.
After calling sanitized_headers on valid headers, the returned dictionary has the Authorization key’s value as <redacted>.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Design explanation question: Why must the scope check come after the expiry check? Explain how this order affects security and performance.

A strong answer should include

Expiry check is a precondition: if the token is expired, it cannot be used regardless of scope, so raising an expiry error first avoids unnecessary scope checks.
If scope were checked first, it might leak scope information or waste computational resources for an expired token.
The order ensures that authorization decisions are only made when the token is valid, aligning with the principle of least privilege.

Mastery evidence

AccessToken.headers_for raises PermissionError when the token is expired or missing the required scope, and the test test_missing_scope_is_rejected passes.
sanitized_headers replaces the Authorization header value with <redacted>, and the test test_authorization_is_redacted_from_logs passes.
In the music scenario transfer challenge, scope rejection and header redaction are successfully verified.

STAGE 06

Retry only safe requests with backoff and bind write requests to idempotency keys

Start with a concrete problem

Why can’t we retry all requests?

When integrating with external SaaS services, network jitter or transient server errors are common, so we might want to automatically retry failed requests.

But not all requests are safe to retry: for example, a POST request that creates an order, if the first request succeeded but the response was lost, retrying would create two orders.

In this stage we will implement a plan_request function that decides retry policy based on HTTP method and whether an idempotency key is provided, to avoid duplicate side effects.

Currently there is no forge/mcp/saas.py file, so tests test_get_retries_are_bounded and test_post_retry_requires_idempotency_key fail with an import error.

Make a prediction

Before implementing plan_request, predict: for plan_request("POST", 1) (without providing an idempotency key), what should the function return?

Return a RequestPlan with attempts equal to 2
Raise ValueError indicating an idempotency key is required
Return a RequestPlan with attempts equal to 1
Return None

Reasoning guide: The correct answer is to raise ValueError. Because POST requests may have side effects, retrying without an idempotency key is unsafe, so the function must refuse to plan a retry.

What it is

Retry policy planner

plan_request is a pure function that computes a RequestPlan dataclass based on HTTP method, transient failure count, and an optional idempotency key.

The function only decides whether to retry and how many attempts to make; it does not perform actual network requests or execute retries.

For GET requests, because they are idempotent, it can safely increase attempts based on transient failures, but attempts are capped.

For POST requests, retries are allowed only if an idempotency key is provided; otherwise it raises ValueError.

What it is not

Not a network client or executor

plan_request does not send HTTP requests or handle responses; it only generates a plan object.

It does not generate idempotency keys or validate their format; it only checks for their presence.

It is not a generic retry decorator that automatically wraps arbitrary functions for retrying.

How it relates to neighboring concepts

Relationship with other components

The RequestPlan dataclass contains method, attempts, and headers fields, where headers includes Idempotency-Key if an idempotency key is provided.

Later SaaS clients can use RequestPlan to actually execute requests and retry according to attempts.

plan_request relies on the caller to pass correct arguments; it maintains no state itself.

Boundary decision for this stage: Boundary decision: plan_request only plans retry policy, not actual network calls or retry execution; for write requests, it must require the caller to provide an idempotency key, otherwise refuse to plan.

Experiment: mcp-saas-streaming-06-idempotent-saas-retries
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/mcp-saas-streaming/06-idempotent-saas-retries/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how plan_request rejects POST retry without idempotency key: call plan_request("POST", 1), observe ValueError.

Suppose we are implementing plan_request. First handle GET requests: for plan_request("GET", 9), because GET is idempotent, we can compute attempts based on transient failures.

Using the formula attempts = 1 + min(transient_failures, 2), when transient_failures=9, min(9,2)=2, so attempts=3.

For POST requests, if no idempotency key is provided, the function should raise ValueError, because retrying a write operation may cause duplicate side effects.

Implementation steps

Create forge/mcp/saas.py file, define RequestPlan dataclass and plan_request function.
In plan_request, convert method to uppercase and check if it is a supported method (GET or POST).
If method is POST and no idempotency key is provided, raise ValueError.
For GET requests, compute attempts as 1 + min(transient_failures, 2).
For POST requests with idempotency key, also compute attempts and include Idempotency-Key in headers.

Verified solution code: forge/mcp/saas.py

from dataclasses import dataclass

@dataclass(frozen=True)
class RequestPlan:
    method: str
    attempts: int
    headers: dict[str, str]

def plan_request(method: str, transient_failures: int, idempotency_key: str | None = None) -> RequestPlan:
    method = method.upper()
    if method not in {"GET", "POST"}:
        raise ValueError("unsupported method")
    if method == "POST" and not idempotency_key:
        raise ValueError("write retries require an idempotency key")
    retryable = method == "GET" or idempotency_key is not None
    attempts = 1 + min(transient_failures, 2) if retryable else 1
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    return RequestPlan(method, attempts, headers)

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Why this implementation

This example shows plan_request rejecting unsafe retries: when method is POST and no idempotency key, the function raises ValueError.

In this way, we force callers to provide an idempotency key when planning write request retries, avoiding duplicate side effects.

Guided practice

Add idempotency key check for POST in plan_request to make test test_post_retry_requires_idempotency_key pass.

Open forge/mcp/saas.py file.
In plan_request function, add method check: if method is not GET or POST, raise ValueError.
Add idempotency key check for POST: if method is POST and idempotency_key is None, raise ValueError.
For GET requests, compute attempts as 1 + min(transient_failures, 2).
For POST requests with idempotency key, also compute attempts and include Idempotency-Key in headers.
Run tests python -m pytest -q tests/test_stage.py to verify.

Progressive hints

Hint 1: Use if method == "POST" and not idempotency_key: to check.
Hint 2: Attempts formula is 1 + min(transient_failures, 2).
Hint 3: headers dict includes "Idempotency-Key": idempotency_key when idempotency key is provided.

Success criteria

plan_request("GET", 9) returns attempts == 3.
plan_request("POST", 1) raises ValueError with error message containing "idempotency".

Fault injection and diagnosis

In the fault lab, we provide a defective forge/mcp/saas.py implementation that does not check idempotency key for POST requests, causing test test_post_retry_requires_idempotency_key to fail. Running python -m pytest -q ../tests/test_stage.py::test_post_retry_requires_idempotency_key shows AssertionError: unsafe POST retry was planned.

In this stage we learned how to distinguish safe and unsafe requests and enforce idempotency keys for write requests. This fault is exactly due to the missing check, so we can fix it by adding the idempotency key check.

Verified fault contract

Target test: test_post_retry_requires_idempotency_key
Faulty file: forge/mcp/saas.py

Diagnosis order

Run test python -m pytest -q ../tests/test_stage.py::test_post_retry_requires_idempotency_key and observe failure.
Inspect plan_request function in forge/mcp/saas.py and notice it does not check idempotency key for POST.
Compare expected behavior: POST without idempotency key should raise ValueError.
Locate defect: function computes retries for all methods without distinguishing GET and POST.

Fix principle: In plan_request function, add idempotency key check for POST: if method is POST and idempotency_key is None, raise ValueError. Also raise ValueError for unsupported methods.

python -m pytest -q ../tests/test_stage.py::test_post_retry_requires_idempotency_key

Transfer challenge

In the office research scenario, you need to use plan_request to plan a POST request with an idempotency key and verify the returned RequestPlan includes the Idempotency-Key header. Also, attempt without a key to raise ValueError.

Changed condition: Change the scenario from generic SaaS calls to a document upload request in office research, using a different idempotency key value.

Write a test or script that calls plan_request("POST", 2, idempotency_key="office-upload-123"), asserts the returned RequestPlan has attempts equal to 3 and headers contains {"Idempotency-Key": "office-upload-123"}. Then call plan_request("POST", 2) and assert it raises ValueError.

Success criteria

For POST with idempotency key, the returned RequestPlan has attempts equal to 3 and headers contains the correct Idempotency-Key.
For POST without idempotency key, it raises ValueError with error message containing "idempotency".

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Design explanation question: Why are GET requests safe to retry while POST is not? Explain how an idempotency key prevents duplicate side effects.

A strong answer should include

GET requests are idempotent, meaning multiple executions do not change server state, so they can be safely retried.
POST requests typically create resources or perform actions that may have side effects, so repeating them can cause duplicate creations or operations.
An idempotency key is a unique identifier that the server can use to recognize duplicate requests and execute the operation only once, preventing duplicate side effects.

Mastery evidence

plan_request returns bounded attempts for GET: plan_request("GET", 9).attempts == 3.
plan_request raises ValueError for POST without idempotency key: plan_request("POST", 1) raises an exception containing "idempotency".
Tests test_get_retries_are_bounded and test_post_retry_requires_idempotency_key both pass.

STAGE 07

Negotiate client/server capabilities and resume compatible sessions only from acknowledged cursors

Start with a concrete problem

Protocol version can be silently changed during session resume

The file forge/mcp/session.py does not exist yet, so tests test_negotiation_uses_intersection and test_resume_rejects_protocol_change fail at import time because the module forge.mcp.session is missing.

Even after adding the module, if the resume function unconditionally trusts the protocol version passed by the caller, a session originally negotiated as 2026-01 could be resumed as 2025-06, causing subsequent stream processing to use the wrong message format.

This stage requires implementing both negotiate and resume, ensuring that on resume the protocol version must match the established session, otherwise a ValueError is raised.

Make a prediction

Before implementing resume, which behavior do you think is correct?

During resume, the protocol version can be renegotiated as long as the client supports it.
Resume must use the original session’s protocol version; any mismatch should be rejected.
Resume only needs to check that the cursor is greater than the acknowledged sequence; version is irrelevant.
Resume should automatically select the latest version supported by both client and server.

Reasoning guide: The correct option is the second one: resume must use the original session’s protocol version. Because the session is already established, the protocol version is a contract between both parties; changing the version means the message format may be incompatible and must be rejected.

What it is

Essence of session negotiation and resume

Session negotiation (negotiate) determines a common protocol version and a set of mutually supported capabilities between client and server, producing an immutable Session object.

Session resume (resume) continues processing events after a disconnection using an acknowledged cursor, but must guarantee the protocol version is unchanged and the cursor does not regress.

What it is not

Session resume is not renegotiation

Resume is not renegotiation: it cannot change the protocol version or capability set, otherwise it would break the established communication contract.

Resume also does not handle actual stream transport or reconnection; it only checks state and returns a new session object.

How it relates to neighboring concepts

Relationships between components

negotiate produces a Session, and resume consumes a Session and returns an updated Session.

The protocol_version and capabilities of a Session must remain unchanged during resume; only acknowledged_sequence may be updated.

Boundary decision for this stage: On resume, if the protocol version is inconsistent or the cursor is less than the acknowledged sequence, a ValueError must be raised rather than silently accepting or attempting renegotiation.

Experiment: mcp-saas-streaming-07-session-negotiation-resume
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/mcp-saas-streaming/07-session-negotiation-resume/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how resume rejects protocol version change: create a 2026-01 session, attempt to resume with 2025-06, observe ValueError.

Assume we already have a negotiated session with protocol version 2026-01, capabilities {"tools"}, and acknowledged sequence 8.

After a disconnection, the client attempts to resume the session with protocol version 2025-06 and cursor 9.

The correct implementation should check protocol version consistency, detect the mismatch, and raise ValueError containing the word “version”.

Implementation steps

Create Session("2026-01", frozenset({"tools"}), acknowledged_sequence=8).
Call resume(session, "2025-06", 9).
Assert that a ValueError is raised and the error message contains “version”.

Verified solution code: forge/mcp/session.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Session:
    protocol_version: str
    capabilities: frozenset[str]
    acknowledged_sequence: int = 0

def negotiate(client_versions: tuple[str, ...], server_versions: tuple[str, ...], client_caps: set[str], server_caps: set[str]) -> Session:
    common = [version for version in server_versions if version in client_versions]
    if not common:
        raise ValueError("no compatible protocol version")
    return Session(common[0], frozenset(client_caps & server_caps))

def resume(session: Session, protocol_version: str, cursor: int) -> Session:
    if protocol_version != session.protocol_version:
        raise ValueError("protocol version changed")
    if cursor < session.acknowledged_sequence:
        raise ValueError("resume cursor would replay acknowledged events")
    return Session(session.protocol_version, session.capabilities, cursor)

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Why this implementation

The resume function first compares the passed protocol_version with session.protocol_version; if they differ, it raises ValueError("protocol version changed").

Thus the test’s except ValueError branch is triggered, and the assertion that the error message contains “version” succeeds.

Guided practice

Add protocol version consistency check in resume to make test test_resume_rejects_protocol_change pass.

Create the file forge/mcp/session.py.
Define the Session dataclass with fields protocol_version, capabilities, and acknowledged_sequence.
Implement negotiate: compute common versions, raise ValueError if none; return a Session with the intersection of capabilities.
Implement resume: check protocol version consistency, raise ValueError if mismatch; check cursor regression, raise ValueError if cursor < acknowledged_sequence; otherwise return a new Session.

Progressive hints

Use if protocol_version != session.protocol_version: for the check.
The error message should contain “version” to pass the test assertion.
Cursor check uses if cursor < session.acknowledged_sequence:.

Success criteria

test_negotiation_uses_intersection passes.
test_resume_rejects_protocol_change passes.

Fault injection and diagnosis

The defect in the failure fixture fixtures/failure.json is that the resume function has been replaced with unconditional session reconstruction, ignoring the protocol version check. When test test_resume_rejects_protocol_change runs, it expects resume to raise ValueError on version mismatch, but instead it returns a new session, causing the test to fail.

The core mechanism learned in this stage is exactly the protocol version consistency check. By implementing the version comparison in resume, the defect can be fixed and the test will pass.

Verified fault contract

Target test: test_resume_rejects_protocol_change
Faulty file: forge/mcp/session.py

Diagnosis order

Run python -m pytest -q ../tests/test_stage.py::test_resume_rejects_protocol_change and observe the test failure.
Inspect the resume function in forge/mcp/session.py and notice it directly returns Session(protocol_version, session.capabilities, cursor) without a version check.
Compare with expected behavior: it should check protocol_version != session.protocol_version and raise ValueError.

Fix principle: Add a version consistency check at the beginning of resume: if protocol_version != session.protocol_version: raise ValueError("protocol version changed"). Also keep the cursor check.

python -m pytest -q ../tests/test_stage.py::test_resume_rejects_protocol_change

Transfer challenge

In the office research scenario, the client supports protocol versions 2026-01 and 2025-06, the server supports 2026-01; client capabilities are {"tools", "resources"}, server capabilities are {"tools", "prompts"}. After using negotiate, verify that the returned Session has protocol version 2026-01 and capabilities {"tools"}. Then attempt to resume that session with 2025-06, expecting a ValueError.

Changed condition: Change the client and server version and capability sets to the specific values of the office research scenario, while keeping the negotiation logic unchanged.

Write a test or script that calls negotiate(("2026-01", "2025-06"), ("2026-01",), {"tools", "resources"}, {"tools", "prompts"}), asserts the returned session has protocol version 2026-01 and capabilities frozenset({"tools"}); then calls resume(session, "2025-06", 0) and asserts a ValueError is raised.

Success criteria

Negotiation result has protocol version 2026-01 and capabilities frozenset({"tools"}).
Resuming with a different version raises ValueError with error message containing “version”.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Design explanation question: Why must protocol version be checked on resume? What concrete consequences would arise if version changes were allowed?

A strong answer should include

Protocol version determines message serialization format and field semantics; version mismatch leads to parsing errors or data loss.
Changing version on resume is a unilateral contract modification; the other party cannot correctly interpret messages, potentially causing security vulnerabilities or state inconsistency.

Mastery evidence

negotiate returns a Session with common protocol version and capability intersection; test test_negotiation_uses_intersection passes.
resume raises ValueError on protocol version mismatch or cursor regression; test test_resume_rejects_protocol_change passes.
In the office research transfer challenge, negotiation succeeds and version-mismatched resume is rejected.

Complete the chapter

Connect Forge to external services through MCP, with a different tool in each scenario.

FORGE / LOCAL CHECK

Local lab self-check

Not started
2Reading
3Lab downloaded
4Test result read
5Local check passed

verification.json is parsed only in this browser and is never uploaded. This checks the lab contract; it is not server certification or third-party endorsement.

Choose verification.json

Primary layer

Depends on

Provides to

Governed by

Concept calibration

MCP

Tool

Harness

MCP Servers, SaaS & Streaming Protocols

Start with a concrete problem

Why is checking only required fields not enough?

Make a prediction

What it is

Tool schema is a contract for argument shape

What it is not

Tool schema is not a security boundary

How it relates to neighboring concepts

Relationship between schema, Harness, and Agent Loop

Start with a concrete problem

How can we prevent resource registration from escaping allowed URI prefixes?

Make a prediction

What it is

What ResourceRegistry is

What it is not

What ResourceRegistry is not

How it relates to neighboring concepts

Relationships with other components

Start with a concrete problem

Missing variables are silently replaced with empty strings, producing incomplete prompts

Make a prediction

What it is

PromptTemplate is a variable contract and renderer

What it is not

PromptTemplate does not handle prompt content quality or model invocation

How it relates to neighboring concepts

Relationship between PromptTemplate and ResourceRegistry

Start with a concrete problem

Stream events must arrive in order and stop after termination

Make a prediction

What it is

What StreamAssembler is

What it is not

What StreamAssembler is not

How it relates to neighboring concepts

Relationships with other components

Start with a concrete problem

A read-only token is misused for a write operation

Make a prediction

What it is

AccessToken is a minimal authorization decision unit

What it is not

AccessToken does not handle token acquisition or refresh

How it relates to neighboring concepts

Order relationship between expiry and scope checks

Start with a concrete problem

Why can’t we retry all requests?

Make a prediction

What it is

Retry policy planner

What it is not

Not a network client or executor

How it relates to neighboring concepts

Relationship with other components

Start with a concrete problem

Protocol version can be silently changed during session resume

Make a prediction

What it is

Essence of session negotiation and resume

What it is not

Session resume is not renegotiation

How it relates to neighboring concepts

Relationships between components