ACCP Proof-of-Concept Codec

A pure-Python reference implementation of the Agent Context Compression Protocol (ACCP) — encoder, decoder, session state, schema registry, domain profiles, and harness adapter contract. No external dependencies required.

Requirements

Python 3.10+
No third-party packages needed
Optional: pytest for structured test output

Project Layout

poc/
├── codec.py        # Encoder (7-step) + Decoder (7-step) + typed value parser + frame builder
├── registry.py     # Schema registry, error taxonomy, key abbreviation maps, hash negotiation
├── session.py      # Hot/warm/cold state tiers, tokenizer profiles, TTL, audit log, checkpoints
├── profiles.py     # Domain profile builders: chat, tool/MCP, transaction, streaming, workflow
├── adapter.py      # AccpAdapter base class — harness integration contract
├── test_codec.py   # 64-test suite covering all features
└── README.md       # This file

Running the Tests

Without pytest (built-in runner)

# from the repo root
python poc/test_codec.py

Expected output:

PASS  serialize 42
PASS  serialize 3.14
PASS  serialize True
...
PASS  adapter sync frame
PASS  adapter send captured
PASS  adapter error frame
PASS  parse spec frame research>done
PASS  parse spec frame planner>req
...

64/64 passed  (0 failed)

With pytest

pip install pytest
pytest poc/test_codec.py -v

Quick-Start Usage

1. Encode a message

from poc.codec import AccpEncoder, AgentMessage
from poc.registry import SchemaRegistry
from poc.session import AccpSession

reg = SchemaRegistry()
session = AccpSession()
encoder = AccpEncoder(reg)

msg = AgentMessage(
    agent_id="planner",
    intent="req",
    operation="schedule",
    payload={"who": "@dev_team", "when": "sprint_14", "priority": "high"},
)

frame = encoder.encode(msg, session)
print(frame.raw)
# → @planner>req:schedule{pri:high|when:sprint_14|who:@dev_team}[mid:...,seq:1,ts:...]

Every encoded frame automatically includes the standard envelope (mid, seq, ts). Keys are abbreviated and fields are sorted canonically.

2. Decode a frame

from poc.codec import AccpDecoder
from poc.registry import SchemaRegistry
from poc.session import AccpSession

session = AccpSession()
decoder = AccpDecoder(SchemaRegistry())

raw = "@research>done:analyze{d:q3_sales|f:[rev:-12%QoQ,churn:+3.2%]}[mid:abc123,seq:1,ts:1714000000]"
msg = decoder.decode(raw, session)

print(msg.agent_id)              # research
print(msg.intent)                # done
print(msg.payload["data"])       # q3_sales
print(msg.payload["findings"])   # ['rev:-12%QoQ', 'churn:+3.2%']
print(msg.envelope.msg_id)       # abc123

3. Domain profiles — the recommended API

Instead of constructing AgentMessage directly, use the profile builders:

from poc import profiles
from poc.codec import AccpEncoder
from poc.registry import SchemaRegistry
from poc.session import AccpSession

enc = AccpEncoder(SchemaRegistry())
session = AccpSession()

# Chat
msg = profiles.chat("assistant", "Here are the Q3 findings.", turn=2)

# Tool / MCP call
msg = profiles.tool_call("orchestrator", "web_search",
                          {"query": "ACCP protocol", "max_results": 5},
                          correlation_id="cid_001")

# Transaction
msg = profiles.transaction("payments", "txn_001", amount=142.50, account="acct_9876")

# Streaming chunk
msg = profiles.stream_chunk("streamer", "infer", chunk_index=0,
                             total_chunks=3, data="Hello", correlation_id="stream_abc")

# Workflow task
msg = profiles.workflow("planner", "Implement auth module",
                         assignee="@dev", priority="high", deadline="sprint_14")

frame = enc.encode(msg, session)
print(frame.raw)

Schema defaults are omitted automatically and restored on decode — no token waste.

4. Typed values

The codec handles Python types natively:

msg = AgentMessage("agent", "req", "op", payload={
    "count":   42,          # → "42"      (integer)
    "score":   3.14,        # → "3.14"    (decimal, canonical)
    "active":  True,        # → "true"    (boolean)
    "note":    None,        # → "~"       (null)
    "tags":    ["a","b"],   # → "[a,b]"   (array)
    "meta":    {"k": 1},    # → "{k:1}"   (map, keys sorted)
    "ref":     "$warm.x",   # → "$warm.x" (reference, resolved on decode)
})

5. Error frames

from poc.codec import build_error_frame, AccpEncoder
from poc.registry import SchemaRegistry, ErrorCode
from poc.session import AccpSession

enc = AccpEncoder(SchemaRegistry())
session = AccpSession()

frame = build_error_frame(
    agent_id="agent",
    error_code=ErrorCode.TOOL_EXEC_FAILED,
    message="web_search timed out",
    source_msg_id="abc123",
    session=session,
    encoder=enc,
)
print(frame.raw)
# → @agent>fail:error{code:E4002|msg:web_search_timed_out|retry:true|schema:ER}[mid:...,seq:1,ts:...]

6. Streaming frames

from poc.codec import build_stream_frame, AccpEncoder
from poc.registry import SchemaRegistry
from poc.session import AccpSession

enc = AccpEncoder(SchemaRegistry())
session = AccpSession()
cid = "stream_001"

chunks = ["Hello", " world", "!"]
for i, chunk in enumerate(chunks):
    frame = build_stream_frame(
        "streamer", "infer", chunk_index=i,
        total_chunks=len(chunks), data=chunk,
        is_final=(i == len(chunks) - 1),
        correlation_id=cid, session=session, encoder=enc,
    )
    print(frame.raw)

7. Policy hooks

from poc.codec import AccpDecoder, AgentMessage
from poc.registry import SchemaRegistry

def block_external_refs(msg: AgentMessage) -> AgentMessage:
    for v in msg.payload.values():
        if isinstance(v, str) and v.startswith("$cold."):
            raise PermissionError(f"Cold state access denied: {v}")
    return msg

decoder = AccpDecoder(SchemaRegistry())
decoder.add_policy_hook(block_external_refs)

8. Audit log

from poc.session import AccpSession

session = AccpSession()
# ... encode/decode messages ...
session.dump_audit()
# [14:03:01] ENCODE msg=49679033 planner>req:schedule
#            frame   : @planner>req:schedule{...}[mid:49679033,seq:1,ts:...]
#            expanded: {'who': '@dev_team', 'when': 'sprint_14', 'priority': 'high'}

9. Tokenizer profiles

from poc.session import set_tokenizer_profile

set_tokenizer_profile("gpt")     # GPT-4o BPE approximation
set_tokenizer_profile("claude")  # Claude (default)
set_tokenizer_profile("gemini")  # Gemini
set_tokenizer_profile("generic") # Conservative fallback

10. Harness adapter

Subclass AccpAdapter to integrate ACCP into any framework:

from poc.adapter import AccpAdapter
from poc.codec import AccpFrame, AgentMessage

class MyFrameworkAdapter(AccpAdapter):
    def on_message_out(self, frame: AccpFrame) -> None:
        my_framework.send(frame.raw)  # write to socket / queue / MCP

    def on_message_in(self, msg: AgentMessage, raw: str) -> None:
        my_framework.route(msg)       # dispatch to agent logic

    def on_tool_call(self, msg: AgentMessage) -> AgentMessage:
        result = my_framework.execute_tool(
            msg.payload["tool_name"], msg.payload["arguments"]
        )
        return profiles.tool_result(self.agent_id, msg.payload["tool_name"],
                                     result, correlation_id=msg.envelope.msg_id)

adapter = MyFrameworkAdapter("my_agent")
adapter.on_session_start()   # emits sync:registry frame
adapter.send(profiles.chat("my_agent", "Hello"))

11. Session state and checkpoints

from poc.session import AccpSession

session = AccpSession()

# Auto-checkpoints fire when hot state exceeds 400 tokens (HOT_CHECKPOINT_TRIGGER)
# Manual checkpoint:
ckpt_id = session.checkpoint()
print(ckpt_id)                    # ckpt_1
print(session.hot_token_count())  # near-zero after checkpoint

12. Custom schema registration

from poc.registry import SchemaRegistry, AccpSchema

reg = SchemaRegistry()
reg.register("incident_report", AccpSchema(
    code="IR", version=1, profile="generic",
    fields=["severity", "affected_service", "owner", "status"],
    defaults={"status": "open", "severity": "medium"},
))
# Use schema_code="IR" on AgentMessage — defaults omitted on encode, restored on decode

13. Registry hash negotiation

from poc.registry import SchemaRegistry

reg = SchemaRegistry()
print(reg.version())   # 2
print(reg.hash())      # e.g. f8586c

# Session start handshake (emitted automatically by adapter.on_session_start()):
# @orchestrator>sync:registry{hash:f8586c|version:2}[mid:...,seq:1,ts:...]

Built-in Schemas

Code	Profile	Key Fields	Notable Defaults
`SR`	generic	period, revenue, growth_pct, segments	period=quarterly
`TA`	workflow	assignee, task, priority, deadline, deps	priority=medium, deps=[]
`CH`	chat	role, content, turn, lang, reply_to	role=assistant, lang=en
`TC`	tool	tool_name, arguments, result, status, error_code	status=ok
`TX`	transaction	transaction_id, amount, currency, account, status	currency=USD, status=pending
`ST`	stream	chunk_index, total_chunks, data, is_final	is_final=false
`ER`	error	error_code, message, retryable, backoff_ms	retryable=false

Standard Envelope Fields

Every frame includes these in its metadata block [...]:

Abbrev	Full	Required
`mid`	msg_id	Always
`seq`	sequence	Always
`ts`	timestamp	Always
`cid`	correlation_id	When replying
`aid`	causation_id	Optional
`sid`	session_id	Recommended
`ttl`	time_to_live	When relevant

Key Abbreviations (selected)

Short	Full	Short	Full
`d`	data	`mid`	msg_id
`f`	findings	`cid`	correlation_id
`pri`	priority	`seq`	sequence
`src`	source	`tool`	tool_name
`dst`	destination	`args`	arguments
`err`	error	`res`	result
`fmt`	format	`txn`	transaction_id
`stat`	status	`amt`	amount
`idx`	chunk_index	`tot`	total_chunks
`code`	error_code	`msg`	message

Full table: registry.py → ABBREV_TO_FULL

Valid Intent Codes

req · done · fail · wait · esc · comp · sync · qry · ack · cancel · stream · end · ping · pong

See protocol-spec.md §4.1 for full descriptions.

Error Code Reference

Code	Name	Retryable
E1001	PARSE_ERROR	No
E1002	INVALID_INTENT	No
E2001	REF_NOT_FOUND	No
E2002	REF_EXPIRED	No
E3001	TIMEOUT	Yes
E3002	DUPLICATE	No
E4001	TOOL_NOT_FOUND	No
E4002	TOOL_EXEC_FAILED	Yes
E5001	POLICY_DENIED	No
E9999	INTERNAL_ERROR	Yes

Full list: registry.py → ErrorCode

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ACCP Proof-of-Concept Codec

Requirements

Project Layout

Running the Tests

Without pytest (built-in runner)

With pytest

Quick-Start Usage

1. Encode a message

2. Decode a frame

3. Domain profiles — the recommended API

4. Typed values

5. Error frames

6. Streaming frames

7. Policy hooks

8. Audit log

9. Tokenizer profiles

10. Harness adapter

11. Session state and checkpoints

12. Custom schema registration

13. Registry hash negotiation

Built-in Schemas

Standard Envelope Fields

Key Abbreviations (selected)

Valid Intent Codes

Error Code Reference

FilesExpand file tree

README.md

Latest commit

History

README.md

File metadata and controls

ACCP Proof-of-Concept Codec

Requirements

Project Layout

Running the Tests

Without pytest (built-in runner)

With pytest

Quick-Start Usage

1. Encode a message

2. Decode a frame

3. Domain profiles — the recommended API

4. Typed values

5. Error frames

6. Streaming frames

7. Policy hooks

8. Audit log

9. Tokenizer profiles

10. Harness adapter

11. Session state and checkpoints

12. Custom schema registration

13. Registry hash negotiation

Built-in Schemas

Standard Envelope Fields

Key Abbreviations (selected)

Valid Intent Codes

Error Code Reference