Forem: Chris Widmer

See SYNAPSE Route a Three-Model Pipeline — No Connector Code Required.

Chris Widmer — Fri, 08 May 2026 20:44:07 +0000

Posts 1 and 2 in this series explained the problem and showed what the connector code looks like before SYNAPSE. This post skips straight to showing it working.

Open the live demo →

What you are looking at

The demo runs a three-model legal document pipeline:

A named entity recognition model extracts parties, jurisdictions, and dates from a contract clause
An obligation classifier assigns each entity a role — licensor, licensee, jurisdiction
A compliance scorer checks the obligations against a GDPR policy

Each model was built by a different team. Each one expects a completely different input format and produces a completely different output format. In a standard pipeline, you would write custom connector code between every model pair — and maintain it every time any model updates its schema.

In the demo, there is no connector code. There are adapter functions.

The one thing to watch for

When Hop 2 appears, look at the left panel — the native input the classifier receives from its ingress adapter.

The NER model produced a field called label. The classifier expects a field called entity_type. Same concept. Completely different names.

The ingress adapter translates between them in four lines, written once. It lives in the adapter, not in a connector file, not in shared pipeline utilities, not in a bridge module that only one person understands. It is part of the model's own interface definition.

When the obligation classifier is updated — when the team that maintains it changes their schema — that translation is updated in the adapter. The scorer downstream never knows anything changed. The NER model upstream never knows anything changed. The canonical IR absorbed it.

What the adapter actually looks like

def ingress(self, ir):
    return [{
        "text":           e["text"],
        "entity_type":    e["label"],   # label → entity_type
        "context_window": ir.payload.content[:80],
        "threshold":      ir.task_header.quality_floor or 0.7,
    } for e in (ir.payload.entities or [])]

That is the complete ingress function for the obligation classifier. It reads from the canonical IR and produces the classifier's native input format. The field name translation happens here and nowhere else.

The provenance chain

After the pipeline completes, the demo shows the full provenance chain — one immutable entry per model, appended in order. Each entry records which model ran, what confidence score it reported, how long it took, and what it cost.

No model can modify a prior entry. The chain is append-only by design. In a production pipeline running HIPAA or GDPR-sensitive data, this chain is your audit trail — automatically maintained by the adapters, without any application code.

Try it yourself

The demo uses pre-computed outputs, but the contract clause is editable. Change the party names or the jurisdiction and re-run — the pipeline logic is the same, the displayed entities update to reflect what you typed.

If you want to go further, the SDK is on PyPI:

pip install synapse-adapter-sdk

The validator will tell you if your adapter is conformant before you register it with any registry:

synapse-validate --adapter my_module.MyAdapter --all-fixtures

Links

This is post 3 in the Building SYNAPSE series. Post 1 covered what MCP solves and what sits above it. Post 2 showed what connector code actually looks like before and after.

The Connector Graveyard: What Multi-Model Pipeline Code Actually Looks Like.

Chris Widmer — Thu, 07 May 2026 08:02:02 +0000

Every ML team building multi-model pipelines has a graveyard.

It is not a literal place. It is a directory called connectors/ or adapters/ or maybe just utils/ with a collection of files named things like ner_to_scorer_bridge.py and v2_classifier_output_transform.py and, if the team has been around long enough, legacy_extractor_compat_DO_NOT_DELETE.py.

Each file solves exactly one problem for exactly one model pair. Each one was written by someone who has since moved on. Each one is a small, undocumented piece of institutional knowledge that the pipeline cannot run without.

This post is about why that graveyard exists, what it actually costs, and what eliminating it looks like in practice.

A concrete pipeline

Let's use a legal document processing pipeline. It is representative of what teams actually build, and the schema incompatibilities are real.

The pipeline needs to do three things in sequence:

Extract named entities (parties, dates, amounts) from an incoming contract
Classify the extracted entities by obligation type
Score the obligations against a compliance policy

Three models. Three teams. Three completely different schemas.

Here is what those schemas actually look like.

Model 1: dslim/bert-base-NER (the extractor)

Input:

{
    "input_text": "The licensor grants a worldwide license to Acme Corp."
}

Output:

[
    {"entity": "B-ORG", "score": 0.9996, "index": 11, "word": "A", "start": 43, "end": 44},
    {"entity": "I-ORG", "score": 0.9449, "index": 12, "word": "##c", "start": 44, "end": 45},
    {"entity": "I-ORG", "score": 0.9990, "index": 13, "word": "##me", "start": 45, "end": 47},
    {"entity": "I-ORG", "score": 0.9994, "index": 14, "word": "Corp", "start": 48, "end": 52}
]

Notice that "Acme Corp" comes back as four wordpiece tokens, not two words. The model uses BERT tokenization. The connector code has to reassemble them.

Model 2: a fine-tuned obligation classifier

Input:

{
    "text": "Acme Corp",
    "entity_type": "ORG",
    "context_window": "grants a worldwide license to Acme Corp",
    "threshold": 0.7
}

Output:

{
    "obligation_type": "LICENSEE",
    "confidence": 0.94,
    "metadata": {
        "model_version": "2.1.0",
        "inference_time_ms": 43
    }
}

Model 3: a compliance scorer

Input:

{
    "obligations": [
        {
            "party": "Acme Corp",
            "role": "LICENSEE",
            "prior_confidence": 0.94
        }
    ],
    "policy_id": "gdpr-commercial-v3",
    "strict_mode": true
}

Output:

{
    "score": 0.87,
    "flags": [],
    "passed": true
}

Three models. Three completely different input shapes. Three completely different output shapes. And none of this is documented anywhere except in the connector code itself.

What the connector code actually looks like

Here is the connector between Model 1 and Model 2. This is realistic code. You have probably written something like it.

def transform_ner_output_to_classifier_input(
    ner_output: list,
    original_text: str,
    confidence_threshold: float = 0.7
) -> list:
    """
    Transform NER output into classifier input format.

    WARNING: NER uses wordpiece tokenization. Must reassemble tokens.
    "Acme Corp" comes back as ["A", "##c", "##me", "Corp"] - see bug 847

    WARNING: Only pass entities above threshold. Classifier chokes on
    low-confidence entities - see incident 2024-11-03
    """
    entities = []
    current_entity = None
    current_tokens = []
    current_start = None
    current_scores = []

    for token in ner_output:
        tag = token["entity"]
        word = token["word"]
        score = token["score"]

        if tag.startswith("B-"):
            if current_entity is not None:
                avg_score = sum(current_scores) / len(current_scores)
                if avg_score >= confidence_threshold:
                    assembled = "".join(
                        t.replace("##", "") for t in current_tokens
                    )
                    start_idx = max(0, current_start - 30)
                    end_idx = min(len(original_text), token["start"] + 30)
                    context = original_text[start_idx:end_idx]

                    entities.append({
                        "text": assembled,
                        "entity_type": current_entity,
                        "context_window": context,
                        "threshold": confidence_threshold
                    })

            current_entity = tag[2:]
            current_tokens = [word]
            current_start = token.get("start", 0)
            current_scores = [score]

        elif tag.startswith("I-") and current_entity is not None:
            current_tokens.append(word)
            current_scores.append(score)
        else:
            current_entity = None
            current_tokens = []
            current_scores = []

    if current_entity is not None:
        avg_score = sum(current_scores) / len(current_scores)
        if avg_score >= confidence_threshold:
            assembled = "".join(t.replace("##", "") for t in current_tokens)
            entities.append({
                "text": assembled,
                "entity_type": current_entity,
                "context_window": original_text[-60:],
                "threshold": confidence_threshold
            })

    return entities

That is 60 lines of connector code to move data between two models. It handles wordpiece reassembly, confidence filtering, context window extraction, and two separate bugs that were discovered in production after the pipeline went live.

Now the connector between Model 2 and Model 3:

def transform_classifier_outputs_to_scorer_input(
    classifier_outputs: list,
    policy_id: str,
    strict_mode: bool = True
) -> dict:
    """
    Aggregate classifier results into scorer input format.

    NOTE: scorer expects "role" not "obligation_type" - same concept, different name
    NOTE: scorer expects "party" not "text" - same concept, different name
    NOTE: prior_confidence is in metadata.confidence in v2.1+, top-level in v2.0

    If classifier returns no obligations above 0.8, scorer returns 422.
    Handle that in the caller. See bug 1089.
    """
    obligations = []

    for item in classifier_outputs:
        if isinstance(item, list):
            result = item[0] if item else None
        elif isinstance(item, dict):
            result = item
        else:
            continue

        if result is None:
            continue

        role = result.get("obligation_type") or result.get("role")
        party = result.get("text") or result.get("party")

        metadata = result.get("metadata", {})
        confidence = metadata.get("confidence") or result.get("confidence", 0.0)

        if role and party:
            obligations.append({
                "party": party,
                "role": role,
                "prior_confidence": confidence
            })

    return {
        "obligations": obligations,
        "policy_id": policy_id,
        "strict_mode": strict_mode
    }

Another 55 lines. Comments explaining that obligation_type and role are the same concept with different names in different schemas. Comments explaining that text and party are the same concept. A note about a 422 error that was discovered in production.

Two connectors. 115 lines of fragile, poorly documented, bug-riddled translation code. And this is a simple three-model pipeline.

What happens when a model updates

In November 2024, the team running the obligation classifier updated it from v2.0 to v2.1. The change was minor: they moved the confidence score from the top level of the response into a metadata object for consistency with their other models.

Before:

{"obligation_type": "LICENSEE", "confidence": 0.94}

After:

{"obligation_type": "LICENSEE", "metadata": {"confidence": 0.94, "model_version": "2.1.0"}}

The connector code was not updated. The prior_confidence field in the scorer input silently defaulted to 0.0 for every obligation. The scorer, receiving obligations with zero confidence, began flagging everything as high risk.

The pipeline did not crash. No exception was raised. The output was wrong for three days before someone noticed the compliance scores had changed. The root cause was a single line in a connector file.

This is not a hypothetical failure mode. It is the most common failure mode in multi-model pipelines. Silent schema drift.

The same pipeline with SYNAPSE adapters

Now look at what the same pipeline looks like with SYNAPSE. Each model has two functions. No connectors. No field name translation logic. No wordpiece reassembly code buried in a utility file.

Adapter for dslim/bert-base-NER:

from synapse_sdk import AdapterBase, CanonicalIR
from typing import Any

class NERAdapter(AdapterBase):
    MODEL_ID = "dslim/bert-base-NER"
    ADAPTER_VERSION = "1.0.0"

    def ingress(self, ir: CanonicalIR) -> dict[str, Any]:
        return {"input_text": ir.payload.content}

    def egress(self, output: list, original_ir: CanonicalIR, latency_ms: int) -> CanonicalIR:
        updated = original_ir.copy()

        entities = []
        current_word = []
        current_label = None
        current_scores = []

        for token in output:
            tag, word, score = token["entity"], token["word"], token["score"]
            if tag.startswith("B-"):
                if current_label:
                    entities.append({
                        "text": "".join(current_word).replace("##", ""),
                        "label": current_label,
                        "confidence": sum(current_scores) / len(current_scores)
                    })
                current_label = tag[2:]
                current_word = [word]
                current_scores = [score]
            elif tag.startswith("I-"):
                current_word.append(word)
                current_scores.append(score)

        if current_label:
            entities.append({
                "text": "".join(current_word).replace("##", ""),
                "label": current_label,
                "confidence": sum(current_scores) / len(current_scores)
            })

        updated.payload.entities = entities
        updated.provenance.append(self.build_provenance(
            confidence=max(e["confidence"] for e in entities) if entities else 0.0,
            latency_ms=latency_ms,
        ))
        return updated

Adapter for the obligation classifier:

class ClassifierAdapter(AdapterBase):
    MODEL_ID = "internal/obligation-classifier-v2"
    ADAPTER_VERSION = "1.0.0"

    def ingress(self, ir: CanonicalIR) -> list[dict]:
        return [
            {
                "text": e["text"],
                "entity_type": e["label"],
                "context_window": ir.payload.content[:60],
                "threshold": ir.task_header.quality_floor or 0.7,
            }
            for e in (ir.payload.entities or [])
        ]

    def egress(self, output: list, original_ir: CanonicalIR, latency_ms: int) -> CanonicalIR:
        updated = original_ir.copy()
        updated.payload.data = {"classified_entities": output}
        updated.provenance.append(self.build_provenance(
            confidence=output[0]["confidence"] if output else 0.0,
            latency_ms=latency_ms,
        ))
        return updated

Adapter for the compliance scorer:

class ScorerAdapter(AdapterBase):
    MODEL_ID = "internal/compliance-scorer-v3"
    ADAPTER_VERSION = "1.0.0"

    def ingress(self, ir: CanonicalIR) -> dict:
        entities = ir.payload.data.get("classified_entities", [])
        return {
            "obligations": [
                {
                    "party": e["text"],
                    "role": e["obligation_type"],
                    "prior_confidence": e["confidence"],
                }
                for e in entities
            ],
            "policy_id": "gdpr-commercial-v3",
            "strict_mode": True,
        }

    def egress(self, output: dict, original_ir: CanonicalIR, latency_ms: int) -> CanonicalIR:
        updated = original_ir.copy()
        updated.payload.data = {"compliance_result": output}
        updated.provenance.append(self.build_provenance(
            confidence=output.get("score", 0.0),
            latency_ms=latency_ms,
        ))
        return updated

What changed

	Connector approach	SYNAPSE adapter approach
Lines of translation code	115 lines across 2 connector files	0 lines of translation code
Field name mapping	Scattered across connectors	Each adapter owns its own mapping
Wordpiece reassembly	In a shared utility, one bug found in prod	In the NER adapter egress, tested against 20 standard fixtures
Schema update impact	Silent failure, 3-day incident	Adapter version bumped, validator catches it before deployment
New model added	New connector file for each existing model	One new adapter file, two functions

When the classifier team updated from v2.0 to v2.1 and moved confidence into the metadata object, the ClassifierAdapter egress function handles that translation once. The canonical IR payload always contains confidence as a normalized field. The ScorerAdapter ingress reads from the canonical IR, not from the classifier's native schema. The schema change in the classifier does not propagate to the scorer because the canonical IR absorbs it.

That is the difference. Not a different architecture. Not a different approach to AI pipelines. Just a shared schema with strict contracts around it, so that schema changes in one model are isolated at the adapter boundary rather than cascading through connector code.

The economics

One well-maintained adapter per model. Written once. Updated when the model's schema changes. Tested against 20 standard fixtures before deployment. Validated against 13 conformance rules.

If you have 5 specialist models in production, you have 5 adapters. Adding a sixth model costs you one adapter, not five new connectors.

If you have 10 models, you have 10 adapters. Not 45 connectors.

The connectors/ directory stays empty. The graveyard does not grow.

If you want to try writing an adapter for a model you are already using, the SDK is at github.com/synapse-ir/adapter-sdk and the specification is at github.com/synapse-ir/spec.

pip install synapse-adapter-sdk
synapse-validate --adapter my_module.MyAdapter --all-fixtures

The BOUNTIES.md lists models where the first adapter earns a Founding Contributor listing.

This is post 2 in the Building SYNAPSE series. Post 1 covered what MCP solves and what it deliberately does not.

What MCP Solves and What It Deliberately Does Not.

Chris Widmer — Thu, 07 May 2026 06:20:15 +0000

MCP is one of the most thoughtfully scoped technical standards to come out of the AI ecosystem. That scope is intentional. Understanding exactly where MCP stops is the precondition for understanding what needs to be built on top of it.

This post is about that boundary.

What MCP actually does

MCP is a JSON-RPC 2.0 protocol that defines three primitives a server can expose: tools, resources, and prompts. It turns the N x M integration problem between LLM applications and external tools into N + M. Build a server once, and any compliant client can use it.

The adoption numbers are striking. In sixteen months, MCP crossed 110 million monthly SDK downloads and more than 10,000 community-built server implementations. Every major AI vendor now supports it. In December 2025, Anthropic donated MCP to the Agentic AI Foundation under the Linux Foundation, co-founded with Block and OpenAI. In April 2026, the AAIF held the MCP Dev Summit North America in New York City with 1,200 attendees, double the previous summit. MCP co-creator David Soria Parra compared MCP's adoption speed to React: React took three years to reach similar numbers. MCP did it in sixteen months.

The mental model that captures it: MCP is the closest thing the LLM ecosystem has to USB-C. A single standard that lets any compliant client talk to any compliant server, without bespoke per-app, per-tool wiring.

A frequently discussed companion protocol is A2A, released by Google in April 2025. The two are complementary, not competing. MCP defines how agents interact with tools. A2A defines how agents collaborate with each other.

So we have the agent-to-tool layer. We have the agent-to-agent layer. Both are under Linux Foundation governance. Both are being adopted at scale.

Here is what neither of them addresses.

The problem they were not designed to solve

Consider a legal AI pipeline that needs to do the following:

Classify an incoming contract by type (NDA, MSA, SOW, license agreement)
Extract all parties, obligations, and termination clauses
Score the extracted obligations against a compliance policy
Generate a risk summary

You have four models. Each was trained by a different team. Each has a different input format, a different output schema, and different field names for the same concepts.

The classifier expects a flat string
The extractor expects a structured JSON object with a document_text field nested under content
The scorer expects an array of obligation objects in a proprietary schema
The summarizer expects a context object with specific metadata fields

None of these models speak to each other natively. MCP does not change this. MCP connects each model to your agent as a tool. What it does not do is define how the output of model A becomes the input of model B when A and B have fundamentally different schemas.

The result: every team building this kind of pipeline writes custom translation code between each model pair.

Models in pipeline	Custom connectors required
4 models	6 connectors
10 models	45 connectors
20 models	190 connectors

Each connector is written once for a specific pair, tested for that pair, and breaks whenever either model updates its schema.

Fivetran's 2026 enterprise data infrastructure benchmark (based on a survey of 500 senior data and technology leaders at organizations with more than 5,000 employees) found that 97% reported pipeline failures slowing their AI programs, with data teams spending 53% of engineering time on pipeline maintenance. That survey covered data pipelines broadly, not AI model pipelines specifically. But anyone who has built multi-model pipelines in production will recognize the pattern.

The maintenance burden is not a people problem. It is a structural problem. There is no shared schema between heterogeneous specialist models, so every connection requires bespoke engineering and every update requires bespoke maintenance.

This is the gap. It is not in MCP's scope, and it should not be. MCP is a connectivity standard. Schema translation between heterogeneous specialist models is a different problem at a different layer.

Why existing tools do not solve this

The existing routing tools solve a related but different problem. OpenRouter routes between general-purpose LLMs: GPT-4o, Claude, Gemini. These models are largely interchangeable at the interface level. You swap one for another and the calling code mostly still works because they all accept similar inputs and produce similar outputs.

Specialist models are not interchangeable. A fine-tuned legal NER model and a clinical ICD coding model have nothing in common at the schema level. Their inputs are different shapes. Their outputs are different shapes. The concepts they operate on have different names in their respective APIs. Routing between them is not a matter of picking a cheaper or faster model for the same task. It is a matter of making fundamentally incompatible systems compose correctly.

Tool	What it solves	What it does not solve
OpenRouter	Cost/quality routing between interchangeable LLMs	Schema translation between specialist models
LiteLLM	Unified gateway for general-purpose LLM APIs	Heterogeneous specialist model interoperability
Martian / NotDiamond	Prompt routing to the right general-purpose model	Domain-specialist model composition
MCP	Agent-to-tool connectivity	How incompatible model outputs become compatible model inputs

None of them address heterogeneous specialist schema translation because they were not built for it.

What a solution looks like

The approach that makes sense is a canonical intermediate representation: a typed, versioned message format that sits between models in a pipeline. Each model gets an adapter with two functions.

ingress converts the canonical format into whatever the model natively expects
egress converts the model's output back into the canonical format and appends a provenance record

Write two functions once per model. The model can now interoperate with every other model in the ecosystem permanently, regardless of what schemas they use natively. Add a new model to your pipeline and you write two functions, not N new connectors.

The adapter contract has to be strict to be useful. The functions must be pure: no network calls inside ingress or egress, no persistent state. The egress function must not modify any provenance record written by a previous model in the chain. The compliance envelope established at the start of the pipeline must be carried through unchanged. These constraints are what make the system composable rather than just connected.

A routing layer on top of this can then make decisions that improve over time. Given a task type, a domain, a latency budget, and a compliance requirement, which registered model has the strongest production track record? That decision gets better as routing signals from real pipeline executions feed back into the routing weights. Declared performance estimates on model cards are a starting point. Observed production performance across many organizations is ground truth.

Why now

At the MCP Dev Summit in April 2026, David Soria Parra said something worth sitting with:

"Behind every corporate firewall, we're quietly wiring MCPs to systems of record and company data all day long. Your Salesforce CRM, your Jira tickets, internal wikis, Snowflake warehouses, HR systems. You never really see or hear about it on Twitter or Hacker News."

That observation applies directly to specialist model pipelines. The integration work is happening, silently, inside every organization that runs more than two specialist models in production. It is being solved independently, in isolation, in ways no one else can reuse.

The developers doing that work are the MCP community's closest neighbors. They already understand protocol layers, adapter patterns, and registry architectures. MCP built that conceptual infrastructure. The moment for a schema translation standard to emerge on top of it is exactly now, while the ecosystem is actively building and the patterns are not yet set.

What we built

We built SYNAPSE: a canonical intermediate representation and adapter framework for heterogeneous AI model pipelines.

The canonical IR is an open specification under CC BY 4.0. The adapter SDK is MIT licensed. The specification covers nine task types, eight domains, a five-layer caching architecture that keeps per-hop overhead under 1ms in steady state, a compliance envelope for GDPR, HIPAA, SOX, and PCI-DSS pipelines, and a calibration layer that improves routing decisions from real production signals over time.

Install the Python SDK:

pip install synapse-adapter-sdk

Writing an adapter looks like this:

from synapse_sdk import AdapterBase, CanonicalIR
from typing import Any

class MyModelAdapter(AdapterBase):
    MODEL_ID = "my-org/my-model-v1.0"
    ADAPTER_VERSION = "1.0.0"

    def ingress(self, ir: CanonicalIR) -> dict[str, Any]:
        return { "input": ir.payload.content }

    def egress(self, output: dict, original_ir: CanonicalIR, latency_ms: int) -> CanonicalIR:
        updated = original_ir.copy()
        updated.provenance.append(self.build_provenance(
            confidence=output["score"],
            latency_ms=latency_ms,
        ))
        return updated

Validate your adapter against 13 conformance rules and 20 standard fixtures:

synapse-validate --adapter my_module.MyModelAdapter --all-fixtures

Links:

Specification: github.com/synapse-ir/spec
Registry server: github.com/synapse-ir/registry
Reference adapters: github.com/synapse-ir/adapters
Docs: synapse-ir.github.io/adapter-sdk

If you are building on top of MCP and running into schema incompatibility between specialist models, this is what SYNAPSE is for. We are actively looking for developers who have the problem and want to write the first community adapters. The BOUNTIES.md lists the models where adapters are most wanted.

SYNAPSE does not compete with MCP. It builds on the layer MCP created.