diff --git a/.editorconfig b/.editorconfig
new file mode 100644
index 0000000..eb79376
--- /dev/null
+++ b/.editorconfig
@@ -0,0 +1,18 @@
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+insert_final_newline = true
+trim_trailing_whitespace = true
+
+[*.py]
+indent_style = space
+indent_size = 4
+
+[{*.yml,*.yaml}]
+indent_style = space
+indent_size = 2
+
+[*.md]
+trim_trailing_whitespace = false
diff --git a/.env.example b/.env.example
index 860697e..46810ef 100644
--- a/.env.example
+++ b/.env.example
@@ -14,6 +14,11 @@
# under the License.
# A2A-T SDK Configuration
+
+# Opt-in full protocol diagnostics. The second switch exposes credentials and cookies;
+# use it only for controlled local troubleshooting.
+WORKFLOW_ENGINE_PROTOCOL_LOGGING=false
+WORKFLOW_ENGINE_PROTOCOL_INCLUDE_SENSITIVE_HEADERS=false
# Copy this file to .env and fill in your values
# Prompt runtime
@@ -35,6 +40,8 @@ A2AT_LLM_BASE_URL=https://api.deepseek.com
A2AT_LLM_MAX_TOKENS=2000
A2AT_LLM_TEMPERATURE=0
A2AT_LLM_TIMEOUT_SECONDS=60
+# none/minimal/low/medium/high/xhigh — set to "none" to disable reasoning on reasoning models
+A2AT_LLM_REASONING_EFFORT=none
A2AT_LLM_HISTORY_WINDOW=10
A2AT_LLM_SESSION_MAX_TOTAL=300
A2AT_LLM_SESSION_MAX_PER_PROVIDER=100
diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 0000000..5e6d123
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1,8 @@
+* text=auto
+
+*.py text eol=lf
+*.md text eol=lf
+*.toml text eol=lf
+*.yml text eol=lf
+*.yaml text eol=lf
+*.json text eol=lf
diff --git a/.github/ISSUE_TEMPLATE/bug_report.yml b/.github/ISSUE_TEMPLATE/bug_report.yml
new file mode 100644
index 0000000..7ca7361
--- /dev/null
+++ b/.github/ISSUE_TEMPLATE/bug_report.yml
@@ -0,0 +1,25 @@
+name: Bug report
+description: Report a reproducible defect in the Python SDK
+title: "[Bug]: "
+labels: [bug]
+body:
+ - type: markdown
+ attributes:
+ value: Do not include credentials, tokens, private AgentCards, or unredacted protocol dumps.
+ - type: input
+ id: version
+ attributes:
+ label: SDK version or commit
+ validations:
+ required: true
+ - type: textarea
+ id: reproduction
+ attributes:
+ label: Minimal reproduction
+ description: Include Python version, expected result, and actual result.
+ validations:
+ required: true
+ - type: textarea
+ id: logs
+ attributes:
+ label: Redacted logs
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
new file mode 100644
index 0000000..1c6b462
--- /dev/null
+++ b/.github/dependabot.yml
@@ -0,0 +1,11 @@
+version: 2
+updates:
+ - package-ecosystem: pip
+ directory: /
+ schedule:
+ interval: weekly
+ open-pull-requests-limit: 5
+ - package-ecosystem: github-actions
+ directory: /
+ schedule:
+ interval: weekly
diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md
new file mode 100644
index 0000000..99469e6
--- /dev/null
+++ b/.github/pull_request_template.md
@@ -0,0 +1,16 @@
+## Summary
+
+Describe what changed and why.
+
+## Compatibility
+
+- [ ] No public API or protocol compatibility impact
+- [ ] Breaking changes are documented with a migration path
+
+## Verification
+
+- [ ] `python -m pytest -q`
+- [ ] `python -m build`
+- [ ] Tests added or updated
+- [ ] Logs and examples contain no credentials or production data
+- [ ] Commit includes a DCO sign-off (`git commit -s`)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..88b96b6
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,29 @@
+name: CI
+
+on:
+ push:
+ branches: [main]
+ pull_request:
+
+permissions:
+ contents: read
+
+jobs:
+ test:
+ strategy:
+ matrix:
+ python-version: ["3.12", "3.13"]
+ runs-on: ubuntu-latest
+ timeout-minutes: 15
+ steps:
+ - uses: actions/checkout@v4
+ - uses: actions/setup-python@v5
+ with:
+ python-version: ${{ matrix.python-version }}
+ cache: pip
+ - name: Install
+ run: python -m pip install --upgrade pip && python -m pip install -e ".[dev]"
+ - name: Test
+ run: python -m pytest -q
+ - name: Build distributions
+ run: python -m build
diff --git a/.github/workflows/dependency-review.yml b/.github/workflows/dependency-review.yml
new file mode 100644
index 0000000..2d9f8ef
--- /dev/null
+++ b/.github/workflows/dependency-review.yml
@@ -0,0 +1,18 @@
+name: Dependency Review
+
+on:
+ pull_request:
+
+permissions:
+ contents: read
+
+jobs:
+ dependency-review:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v4
+ - uses: actions/dependency-review-action@v4
+ with:
+ fail-on-severity: high
+ # Temporary: this repository has not enabled Dependency graph yet.
+ continue-on-error: true
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
new file mode 100644
index 0000000..d94c47d
--- /dev/null
+++ b/.github/workflows/release.yml
@@ -0,0 +1,67 @@
+name: Release to PyPI
+
+on:
+ push:
+ tags:
+ - "v*"
+ workflow_dispatch:
+
+permissions:
+ contents: write
+
+jobs:
+ release:
+ name: Test, publish, and create GitHub release
+ runs-on: ubuntu-latest
+ steps:
+ - name: Checkout
+ uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: "3.12"
+ cache: pip
+
+ - name: Install
+ run: python -m pip install --upgrade pip && python -m pip install -e ".[dev]"
+
+ - name: Validate release version
+ if: startsWith(github.ref, 'refs/tags/v')
+ shell: bash
+ run: |
+ VERSION="${GITHUB_REF_NAME#v}"
+ PROJECT_VERSION="$(python -c 'import tomllib; print(tomllib.load(open("pyproject.toml", "rb"))["project"]["version"])')"
+ if [[ "${VERSION}" != "${PROJECT_VERSION}" ]]; then
+ echo "Tag v${VERSION} does not match pyproject version ${PROJECT_VERSION}"
+ exit 1
+ fi
+
+ - name: Test
+ run: python -m pytest -q
+
+ - name: Build distributions
+ run: python -m build
+
+ - name: Publish distributions
+ if: startsWith(github.ref, 'refs/tags/v')
+ uses: pypa/gh-action-pypi-publish@release/v1
+ with:
+ password: ${{ secrets.PYPI_API_TOKEN }}
+
+ - name: Create GitHub release
+ if: startsWith(github.ref, 'refs/tags/v')
+ shell: bash
+ env:
+ GH_TOKEN: ${{ github.token }}
+ run: |
+ VERSION="${GITHUB_REF_NAME#v}"
+ NOTES="$(awk -v ver="$VERSION" '
+ index($0, "## [" ver "]") == 1 { found = 1; next }
+ found && /^## \[/ { exit }
+ found { print }
+ ' CHANGELOG.md)"
+ gh release view "${GITHUB_REF_NAME}" >/dev/null 2>&1 || \
+ gh release create "${GITHUB_REF_NAME}" --verify-tag \
+ --title "${GITHUB_REF_NAME}" --notes "${NOTES}"
+ gh release upload "${GITHUB_REF_NAME}" dist/* --clobber
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2f88608..d42be39 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,30 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+## [0.0.9] - 2026-09-18
+
+### Changed
+- Align the business callback contract with the Java engine: callbacks prepare final content and never receive the transport client.
+- Replace rendered Markdown predecessor context with typed `WorkflowInput` and ordered task outputs.
+- Evaluate every conditional outgoing edge independently; unconditional edges always pass and a node may activate zero through N successors.
+- Integrate `a2a-t-sdk>=1.0.9,<2` core metadata types while keeping content generation and semantic validation in host code; remove implicit Task-T handlers and legacy negotiation schemas.
+- Install development dependencies directly through the editable dev extras and remove the temporary bootstrap file.
+
+### Added
+- Structured response evidence, protocol-to-business result mapping, safe `BusinessFailure`, and standard A2A error mapping.
+- Remote `get_task`, `list_tasks`, `cancel_task`, and `subscribe_to_task` operations.
+- Explicit `NotificationSubscription` acknowledgement, completion, close, heartbeat, and health lifecycle.
+- mTLS/CRL transport configuration and fail-closed TLS file validation.
+- Reusable per-agent credential profiles with nested overrides and fail-fast configuration checks.
+
+### Fixed
+- Preserve message, task, and artifact metadata at their original levels and process status/artifact stream updates.
+- Detect duplicate graph edges, invalid history sources, duplicate AgentCards, and conflicting authentication headers.
+- Use the current A2A interceptor contract so activated extension and complete authentication requirement headers reach the wire.
+- Close only resources owned by the high-level runner.
+- Read canonical `negotiationContext` metadata and enforce current Propose/Accept/Reject/Abort context and round rules.
+- Serialize protobuf message parts as their JSON fields in workflow events instead of leaking descriptor internals.
+
## [0.0.3] - 2026-08-10
### Fixed
diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md
new file mode 100644
index 0000000..4dc7f6d
--- /dev/null
+++ b/CODE_OF_CONDUCT.md
@@ -0,0 +1,9 @@
+# Code of Conduct
+
+We are committed to a welcoming, harassment-free community. Be respectful,
+focus criticism on the work, and do not disclose private information.
+
+Maintainers may moderate contributions that violate these expectations.
+Reports should be sent privately to the maintainers listed in
+[MAINTAINERS.md](MAINTAINERS.md). Security vulnerabilities must follow
+[SECURITY.md](SECURITY.md).
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 0000000..0e40582
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,30 @@
+# Contributing to workflow-engine-sdk-python
+
+Thank you for contributing. Development requires Python 3.12 or newer.
+
+## Setup
+
+```bash
+python -m venv .venv
+python -m pip install -e ".[dev]"
+python -m pytest -q
+python -m build
+```
+
+Keep Java and Python protocol behavior aligned when changing workflow models,
+events, routing, Task-T, or Negotiation-T. Add regression tests for both SDKs
+when a change affects shared semantics.
+
+Do not commit `.env` files, credentials, tokens, certificates, private
+AgentCards, or production protocol payloads. INFO logs must contain lifecycle
+metadata only; payload diagnostics are opt-in and must be redacted before they
+are attached to an issue or pull request.
+
+Use conventional commit messages and add a DCO sign-off:
+
+```bash
+git commit -s -m "fix: describe the change"
+```
+
+Pull requests should explain motivation, compatibility impact, tests, and any
+required migration. Security reports must follow [SECURITY.md](SECURITY.md).
diff --git a/DESIGN.md b/DESIGN.md
index 30acccc..c3b997a 100644
--- a/DESIGN.md
+++ b/DESIGN.md
@@ -1,358 +1,116 @@
-# A2A-T Engine SDK - Design
+# Workflow Engine Python SDK — Design
-> Architecture and design rationale for the `workflow-engine` Python SDK and the
-> `workflow-engine-java` Java SDK. This document describes the system as shipped
-> in v1.0; the two SDKs are intentionally parallel and maintain feature
-> parity. It is written for engineers integrating or extending the SDK, not
-> as a walkthrough of any particular bug-fix history.
+## 1. Boundary
----
+The engine is protocol-orchestration infrastructure embedded in a host agent. It owns:
-## 1. Overview
+- workflow validation, dependency resolution, parallel scheduling, and lifecycle events;
+- A2A message envelopes, transport selection, authentication headers, task/context identity, remote waiting, and task APIs;
+- detection and coordination of a Negotiation-T exchange after a valid `INPUT_REQUIRED` response;
+- structured projection of remote messages, artifacts, outputs, and protocol failures.
-The A2A-T Engine SDK lets a host agent execute multi-step workflows over the
-A2A protocol with A2A-T telecom extensions. A workflow is a directed
-acyclic graph (DAG) of steps; each step dispatches one or more tasks to
-remote agents and routes to the next step. The SDK owns the protocol
-mechanics - message send, streaming, authentication, Task-T prompt
-generation, the Negotiation-T auto-loop - and exposes a small set of
-decision interfaces the host implements for business logic.
+The host owns:
-The guiding principle is a clean separation between **mechanics the SDK
-owns** and **decisions the host owns**:
+- schemas, templates, LLM calls, A2A-T generation/validation, and business interpretation;
+- final content returned by `on_task` and `on_negotiation`;
+- local work in `on_self_task` and per-edge decisions in `on_route`;
+- when independent Authorization-T and Notification-T operations run and whether their failures matter to host policy.
-| The SDK owns (protocol mechanics) | The host owns (business decisions) |
-|---|---|
-| A2A message send, streaming, SSE normalization | Whether and when to send a task |
-| Agent authentication (Bearer, custom headers) | Credential configuration |
-| A2A-T extensions (Task-T, Negotiation-T, Authorization-T, Notification-T) | Authorization approval, notification handling |
-| DAG traversal, context assembly, state management | Branch routing decisions |
-| Event emission | Event handling |
+The engine depends only on the current `a2a-t-sdk` core metadata model. `A2atMessages` converts typed `MetadataContent` and reads canonical `NegotiationContext`; it does not generate prompts, invoke an LLM, validate business semantics, or use the deprecated state-machine negotiation API. Those content operations remain in host code.
----
+## 2. Architecture
-## 2. Layered Architecture
+```text
+Host agent
+ ├─ ControlPoint (business content and decisions)
+ ├─ WorkflowExecutor (DAG, readiness, parallelism, result aggregation)
+ ├─ WorkflowEngineClient (task/context association, wait, negotiation loop)
+ └─ A2ATransport (A2A SDK, auth, TLS, stream reduction)
-The SDK is structured in four layers. Each layer builds on the one below;
-each has a single responsibility and a clear entry point.
-
-```
-```mermaid
-graph TD
- L2["Layer 2 - Orchestration
execute_psop / ExecutePsop
lifecycle, event stream, cancellation, onFinish persistence"]
- L1["Layer 1 - Traversal
WorkflowExecutor
DAG walk, parallel dispatch, context assembly, routing"]
- L0["Layer 0 - Communication
A2ATransport + two facades
WorkflowEngineClient (workflow send) | ExtensionSender (one-shot)"]
- F["Foundation - Decision
ControlPoint
user-implemented business decisions"]
-
- L2 --> L1 --> L0
- L0 -.-> F
+Independent channels
+ └─ ExtensionSender
+ ├─ Authorization-T one-shot operation
+ └─ NotificationSubscription long-lived stream
```
-### 2.1 Layer 0 - Communication
-
-This layer is the heart of the transport-facade split.
-
-**`A2ATransport`** is the shared wire layer. It owns exactly one concern:
-getting bytes to and from remote agents. That means the HTTP client
-(httpx in Python, the A2A SDK client runtime in Java), the auth manager and
-interceptors, the agent-card map, and the streaming-response consumer. It
-exposes two send primitives - `send` (collect-and-return) and
-`sendNotificationStream` (long-lived SSE) - plus static extractors that turn
-the raw SDK event stream into text, task state, and metadata.
-
-**Two facades sit on top of the transport, each with a single
-responsibility:**
-
-- **`WorkflowEngineClient`** - the workflow execution send path. Owns Task-T
- prompt generation (before send), the Negotiation-T auto-loop (after
- receive), the global `EventCallback`, and the `ControlPoint` wiring.
- This is the facade the executor calls during workflow execution.
-- **`ExtensionSender`** - one-shot pre-positioning. Sends Authorization-T
- and Notification-T messages to agents *before* the workflow starts. It
- bypasses Task-T generation and the negotiation loop and does not emit
- events through the global callback - the returned result *is* the
- callback.
-
-#### Why a shared transport with two facades?
-
-Both the workflow send path and the one-shot pre-positioning path need the
-same wire-level machinery: an HTTP client, TLS configuration, auth
-interceptors, agent-card resolution, and SSE parsing. Putting that machinery
-on either facade would either (a) force a caller that only wants to
-pre-position to hold the full workflow facade, or (b) duplicate the wire code
-across two classes. The shared-transport / two-facade design avoids both:
-the wire layer is written once on `A2ATransport`, and each facade delegates
-all wire work to it while keeping its own orchestration concern isolated.
-
-### 2.2 Layer 1 - Traversal
-
-**`WorkflowExecutor`** walks the DAG. At each step it assembles upstream
-context (`ContextBuilder`), dispatches subtasks concurrently, applies the
-step's success policy, and determines the next step(s). It delegates every
-*decision* to `ControlPoint` and every *send* to `WorkflowEngineClient`.
-
-Step dispatch rules:
-
-- Steps whose predecessors are all satisfied are collected and dispatched in
- parallel (`asyncio.gather` / `CompletableFuture`), so steps at the same
- layer run concurrently.
-- Subtasks within a step also run in parallel.
-- `ALL_SUCCESS` - all subtasks must succeed.
-- `ANY_SUCCESS` - the first successful subtask wins; the rest are cancelled.
-- `SELF_LOOP` - the task is handled locally via `onSelfTask`, with no A2A-T
- message sent to the agent.
-
-### 2.3 Layer 2 - Orchestration
-
-**`execute_psop`** (Python) / **`ExecutePsop`** (Java) is the high-level
-runner. It wraps the executor with a lifecycle (start / complete / error /
-close), event serialization, client-disconnect cancellation, and an
-`onFinish` persistence hook. Most integrations use this layer.
+`WorkflowExecutor` depends on the client abstraction, not on a concrete network. A caller can replace `A2ATransport` with another transport adapter that preserves the same client contract.
----
+## 3. Business callback contract
-## 3. Decision Interfaces
+`ControlPoint` has four callbacks:
-The SDK exposes two user-implemented interfaces, split by responsibility.
+| Callback | Input | Output | Engine action |
+|---|---|---|---|
+| `on_task` | `TaskRequest` | final `MessageContent` | wraps and sends it |
+| `on_self_task` | `TaskRequest` | `TaskResult` | records local outputs; no A2A send |
+| `on_route` | one `RouteRequest` | `RouteDecision` | activates or suppresses that edge |
+| `on_negotiation` | `NegotiationRequest` | `NegotiationSend` or `NegotiationStop` | continues the same remote task or stops locally |
-### 3.1 ControlPoint - flow decisions
+No callback receives a transport client. The engine never treats missing handlers as success, route selection, or negotiation consent.
-Drives the workflow forward. Each method is called by the executor or the
-auto-negotiate loop and makes exactly one decision:
+## 4. Data flow
-| Method | Called by | Decision |
-|---|---|---|
-| `onTask` | executor | Send a task to an agent (call `sendMessage`) |
-| `onSelfTask` | executor | Handle a self-loop task locally (no A2A-T) |
-| `onRoute` | executor | Choose a branch at a conditional step |
-| `onNegotiation` | client auto-loop | Supply clarification on INPUT_REQUIRED |
+`TaskRequest.input` is the current task input. `TaskRequest.workflow_input` is a separate history window selected by the workflow:
-Authorization-T and Notification-T are pre-positioning concerns handled
-via `ExtensionSender` before the workflow starts, not in-workflow callbacks.
-
----
-
-## 4. A2A-T Extension Model
-
-Four A2A-T extensions are supported. They divide into two groups by
-lifecycle.
-
-### 4.1 In-workflow extensions
-
-Participate in every `sendMessage` lifecycle through the extension handler
-chain (`ExtensionRegistry` pre-registers both):
-
-- **Task-T** - On send, calls the A2A-T SDK to generate a structured task
- prompt from the natural-language message and injects it into the message
- metadata. Skipped for negotiation follow-ups and when the caller pre-sets
- the prompt. On receive: pass-through.
-- **Negotiation-T** - On receive, when the agent returns `INPUT_REQUIRED`
- and declares the extension, extracts the negotiation context and message.
- This feeds the auto-loop: the engine calls `ControlPoint.onNegotiation`
- for a clarification, resends the follow-up, and repeats up to a configured
- round limit.
-
-### 4.2 Pre-positioning extensions
-
-One-shot sends that happen before the workflow starts, via `ExtensionSender`:
-
-- **Authorization-T** - Sends an authorization pre-positioning request.
- The prompt value is generated by the A2A-T SDK; the engine falls back to
- the raw natural-language input when SDK generation is unavailable.
-- **Notification-T** - Establishes a result subscription. Opens a long-lived
- SSE stream so later recovery results flow back through the response
- stream.
-
-The subscription *result* (e.g. a recovery outcome pushed later) flows back
-through the `sendNotification` response stream, not through
-`onNotification`. That hook only fires when an agent voluntarily includes a
-Notification-T payload in a `sendMessage` task response.
-
-Prompt generation methods for Authorization-T and Notification-T are
-reserved on `ExtensionSender`; Task-T uses the SDK's `generateTaskPrompt`.
-Negotiation-T and the reserved generators will be wired to the A2A-T SDK as
-that support lands upstream.
-
-### 4.3 Extension handler chain
-
-```mermaid
-graph TD
- SM["sendMessage(agent, message)"]
- BS["before_send: Task-T generates prompt, injects into metadata"]
- TS["transport.send (Task-T metadata on the wire)"]
- AR["after_receive: Negotiation-T extracts context (feeds auto-loop)"]
- AN["auto_negotiate loop (if INPUT_REQUIRED)"]
-
- SM --> BS --> TS --> AR --> AN
+```text
+WorkflowInput
+ runtime_intent
+ upstream_results[]
+ step_name
+ task_results[]
+ agent_name / skill / logical task_id / task_description / status
+ outputs[]
+ received_messages[]
+ error / error_code / error_details
```
-`ExtensionRegistry.getHandlersForExtensions` matches an agent's declared
-extension URIs against handler keywords (case-insensitive) and returns the
-handler chain for that agent. Authorization-T / Notification-T handler
-classes are retained for callers that need inline handling of agent-pushed
-data, but they are not auto-registered - that is a pre-positioning concern.
-
----
-
-## 5. Condition Routing
-
-A step's `next` list holds `JumpCondition(step, condition)` entries. The
-routing rule is:
-
-- **No `next`** - terminal; the step completes the branch.
-- **All conditions empty** - unconditional fan-out: dispatch every
- non-terminal next step in parallel.
-- **Has conditions** - conditional: call `ControlPoint.onRoute`, which
- returns a single `RouteDecision.nextStep`. The engine enforces that the
- returned step is among the declared conditions; an invalid step ends the
- workflow with a warning.
-
-This makes conditional branches an N-choose-1 selection and keeps
-unconditional fan-out as automatic parallel dispatch.
-
----
-
-## 6. Event Model
-
-Events are emitted to an optional `EventCallback` as stable string types
-(`EventType`). They are grouped by origin:
-
-- **Runner lifecycle** - `start`, `complete`, `close`
-- **Step / task execution** - `step_start`, `step_complete`, `task_request`,
- `task_response`, `task_status_changed`, `route_decision`,
- `workflow_complete`
-- **Agent traffic** - `agent_request`, `agent_response`,
- `agent_status_update`, `agent_artifact_update`, `agent_message_event`
-- **A2A-T extensions** - `negotiation_request`, `negotiation_resolved`,
- `negotiation_failed`, `authorization_request`, `authorization_resolved`,
- `notification`
-- **Failure** - `error`, emitted on step failure by the executor and on
- final failure by the runner
-
----
-
-## 7. Interaction Sequences
-
-### 7.1 Workflow execution with negotiation
-
-```mermaid
-sequenceDiagram
- participant H as Host
- participant E as Executor
- participant C as EngineClient
- participant A as Agent
-
- H->>E: run(workflow)
- E->>C: onTask(req)
- C->>A: before_send: Task-T
- A-->>C: INPUT_REQUIRED (Negotiation-T)
- C->>E: negotiation result
- E->>H: onNegotiation (host supplies clarification)
- H->>E: clarification
- E->>C: follow-up send
- C->>A: send follow-up
- A-->>C: final result
- C->>E: final result
- E->>H: ExecutionResult
-```
+Remote A2A messages are reduced to `ReceivedMessage` snapshots. Message metadata, task metadata, and artifact metadata remain at their original levels. Outputs are projected from final message parts when no artifact exists, or from artifacts when they do. Failed status messages remain evidence and do not become business output.
-### 7.2 Pre-positioning authorization
-
-```mermaid
-sequenceDiagram
- participant H as Host
- participant ES as ExtensionSender
- participant T as Transport
- participant A as Agent
-
- H->>ES: sendAuthorization(agent)
- ES->>ES: generate prompt (SDK)
- ES->>T: send(instruction, auth)
- T->>A: send
- A-->>T: auth result
- T-->>ES: auth result
- ES-->>H: result
-```
+## 5. Scheduling and routing
-### 7.3 Notification subscription
-
-```mermaid
-sequenceDiagram
- participant H as Host
- participant ES as ExtensionSender
- participant T as Transport
- participant A as Agent
-
- H->>ES: sendNotification
- ES->>T: sendNotificationStream
- T->>A: open long-lived SSE
- A-->>T: ack (working)
- T-->>ES: first event -> future
- ES-->>H: result
- Note over T,A: later results stream back over the same connection
-```
+All root steps start independently. A step becomes ready after every activated predecessor has produced a result. Ready steps run in parallel; subtasks within a step also run in parallel.
----
+- `ALL_SUCCESS`: all subtasks must succeed.
+- `ANY_SUCCESS`: the first success wins and unfinished sibling calls are cancelled.
+- `SELF_LOOP`: each subtask is handled by `on_self_task` without a network send.
-## 8. Cross-SDK Parity
+Outgoing edges are evaluated as a set:
-The Python and Java SDKs are designed for feature parity. Module-to-class
-mapping:
+- a blank condition is unconditional and allowed without a callback;
+- every nonblank condition invokes `on_route` independently, including mixed conditional/unconditional fan-out;
+- zero, one, or many successors may be activated;
+- all route evaluations finish before any successor is scheduled;
+- one route callback failure suppresses every successor of that source and fails execution;
+- all denied edges end that branch successfully.
-| Concern | Python module | Java class |
-|---|---|---|
-| Shared wire layer | `client/a2a_transport.py` | `client/A2ATransport` |
-| Workflow send facade | `client/engine_client.py` | `client/DefaultWorkflowEngineClient` |
-| One-shot facade | `client/extension_sender.py` | `client/DefaultExtensionSender` |
-| Extension handlers | `client/extension_handlers.py` | `client/TaskTHandler`, `NegotiationTHandler`, `ExtensionRegistry` |
-| Extension enums | `client/extensions.py` | `client/A2ATExtension` |
-| Auth | `client/auth_manager.py`, `credential_service.py` | `client/AgentAuthManager`, `AgentCredentialService` |
-| SSL/TLS | `client/ssl_context.py` | `client/SslContextFactory` |
-| SSE normalization | `client/sse_normalization.py` | `client/SseNormalization` |
-| Flow decisions | `control/control_points.py` | `control/ControlPoint`, `DefaultControlPoint` |
-| Events | `control/control_points.py` (`EventType`) | `control/EventType`, `EventCallback` |
-| DAG traversal | `core/executor.py` | `core/WorkflowExecutor` |
-| Context assembly | `core/context_builder.py` | `core/ContextBuilder` |
-| Models | `core/models.py` | `model/*` |
-| Registry | `registry/registry_client.py` | `registry/RegistryClient`, `LoadPsop` |
-| Runner | `runner.py` (`execute_psop`) | `runner/ExecutePsop` |
+Graph validation rejects blank or duplicate step names, reserved terminal names, missing or duplicate edge targets, cycles, invalid roots, invalid `context_from` entries, and combining `"*"` with named context sources.
-Naming conventions differ by language (Python snake_case, Java camelCase)
-but the public surface, event types, extension URIs, and model fields are
-aligned.
+## 6. Task and negotiation lifecycle
----
+For each logical subtask, the executor generates a stable local task ID. The client generates a fresh A2A context ID, sends host-provided content, and records the first remote task ID. Follow-up sends must preserve both remote IDs.
-## 9. Dependencies
+`SUBMITTED` and `WORKING` are acknowledgements, not final results. The client polls the A2A task until a terminal state or `INPUT_REQUIRED`. `INPUT_REQUIRED` is accepted as Negotiation-T only when the response contains a valid Negotiation-T context. The callback receives the original submission, current response evidence, previous exchanges, and remaining interaction budget.
-**Python SDK:** `a2a-sdk` (A2A protocol), `a2a-t-sdk` (A2A-T extensions),
-`httpx`, `loguru`, `protobuf`, `packaging`.
+The host may send generated continuation or terminal content, or stop locally. The engine validates transport identity and progression but leaves A2A-T semantic validation to the host SDK; it does not invent an Abort. If local stop, timeout, cancellation, or invalid protocol data ends an interaction after a remote task ID is known, the engine makes a bounded best-effort A2A task-cancel call and preserves the original failure. Repeated proposals are deduplicated and observed rather than resending the business command. The content callback, remote execution, waiting, and negotiation exchanges share one timeout budget.
-**Java SDK:** `org.a2aproject.sdk:a2a-java-sdk-client` (A2A protocol),
-`net.openan.a2at.sdk:a2a-t-client` (A2A-T extensions), Jackson, SLF4J,
-Lombok.
+## 7. Independent operations
-Both SDKs are standalone: they do not depend on the orchestration center.
-The orchestration center consumes the Python SDK as a library.
+Authorization-T is a one-shot operation that waits through `SUBMITTED` or `WORKING` until the remote task reaches a final result. Notification-T returns a `NotificationSubscription` before I/O begins, so even an early listener callback can close it. The acknowledgement future exposes the first protocol result; the host must reject failed, canceled, or rejected acknowledgements before treating the subscription as established. The completion future represents the SSE channel lifetime. Event count and last-event time provide local liveness only.
----
+These operations must use channel instances independent from workflow task traffic. Their success is not an engine prerequisite for workflow execution.
-## 10. Design Decisions Summary
+## 8. Failure model
-1. **Shared transport, two facades** - wire machinery written once on
- `A2ATransport`; `WorkflowEngineClient` and `ExtensionSender` each own one
- orchestration concern and delegate wire work. Avoids both forced-facade
- coupling and wire-code duplication.
+The current A2A SDK parses official REST/JSON-RPC/gRPC error structures into `A2AError` subclasses. The engine maps them to stable `a2a.` codes and preserves safe protocol facts. A remote task that was created successfully but later failed remains an HTTP-successful interaction; its failed, rejected, or canceled task state is exposed through stable `a2a.task_*` failure semantics.
-2. **In-workflow vs pre-positioning extensions** - Task-T and Negotiation-T
- are part of the `sendMessage` chain; Authorization-T and Notification-T
- are one-shot sends before the workflow. The registry auto-registers only
- the in-workflow pair.
+Host code may raise `BusinessFailure(code, message, details)` for explicitly safe diagnostics. Generic exceptions use the stable `workflow.execution_failed` code and preserve their message so integration errors remain actionable; credential and provider integrations remain responsible for supplying redacted exception text.
-3. **Auto-negotiation loop** - the engine owns the resend loop so hosts only
- supply clarification text (`onNegotiation`), never the protocol
- mechanics of resending.
+## 9. Resource and security rules
-4. **Condition routing semantics** - empty conditions mean fan-out
- (parallel), conditional branches mean N-choose-1 via `onRoute`. Keeps
- the routing model predictable.
+- TLS server verification defaults to enabled and configured trust/client files fail closed.
+- `ssl_verify=False` is an explicit per-client development opt-out.
+- AgentCard authentication requirements must be satisfiable by configured credentials or `AuthProvider`.
+- credential profiles are resolved per agent and explicit malformed or missing configuration fails during construction.
+- Conflicting header values from credentials and a custom provider fail instead of silently overwriting.
+- caller-created transports and HTTP clients remain caller-owned; high-level `execute_psop` closes only clients it creates.
+- full protocol payload logging is opt-in and sensitive headers are redacted.
diff --git a/DEVELOPER_GUIDE.md b/DEVELOPER_GUIDE.md
index 6601d2e..1450b41 100644
--- a/DEVELOPER_GUIDE.md
+++ b/DEVELOPER_GUIDE.md
@@ -1,516 +1,349 @@
-# Developer Guide: workflow-engine SDK
+# Developer Guide
-Integration guide for the workflow-engine Python SDK: installation, the layered
-entry points, ControlPoint implementation, event handling, agent
-authentication, and the A2A-T extension model. For the architecture
-rationale, see [DESIGN.md](DESIGN.md).
+## 1. Required versions
----
-
-## 1. Installation
+- Python 3.12 or newer
+- `a2a-sdk>=1.1.2,<2`
+- `a2a-t-sdk>=1.0.9,<2` (the engine uses its core metadata contract; host callbacks use its content APIs)
```bash
-pip install workflow-engine
+python -m pip install -e ".[dev]"
+python -m pytest -q
```
-Dependencies (auto-installed): `a2a-sdk`, `a2a-t-sdk`, `httpx`, `loguru`,
-`protobuf`, `packaging`.
+Construct the host's `A2ATClient` with an existing `.env` path. Text-generation and semantic-validation calls need a configured LLM; deterministic negotiation `from_data` generation does not make an LLM call.
-Verify:
+## 2. Workflow model
```python
-import workflow_engine
-print(workflow_engine.__version__) # 0.0.3
+workflow = Workflow.from_dict({
+ "name": "parallel-diagnosis",
+ "steps": [
+ {
+ "name": "dispatch",
+ "layer": 0,
+ "step_type": "AllSuccess",
+ "subtasks": [
+ {"agent": "domain-a", "skill": "diagnose", "description": "diagnose A"},
+ {"agent": "domain-b", "skill": "diagnose", "description": "diagnose B"},
+ ],
+ "next": [{"step": "aggregate", "condition": ""}],
+ },
+ {
+ "name": "aggregate",
+ "layer": 1,
+ "step_type": "SelfLoop",
+ "context_from": ["dispatch"],
+ "subtasks": [{"agent": "host", "description": "aggregate results"}],
+ },
+ ],
+})
```
----
-
-## 2. Core Concepts
+`Task.input` may contain natural-language text or JSON-compatible structured data:
-The SDK has three layers. Pick the one that matches your use case:
+```python
+Task(agent="domain-a", input=BusinessInput.from_text("diagnose circuit"))
+Task(agent="domain-b", input=BusinessInput.from_data({"circuit_id": "C-1"}))
+```
-| Layer | Entry Point | What It Handles | What You Provide |
-|-------|------------|-----------------|-----------------|
-| **2 (high)** | `execute_psop()` | Event stream, lifecycle, cancellation, event collection | ControlPoint + AgentCards + config |
-| **1 (mid)** | `WorkflowExecutor` | DAG traversal, context assembly, ControlPoint dispatch | ControlPoint + WorkflowEngineClient + Workflow |
-| **0 (low)** | `A2ATransport` + facades | A2A send, auth, extensions, SSE normalization | AgentCards + config |
+When omitted, the task description becomes the text input.
-**Most integrations use Layer 2 (`execute_psop`).** It handles the lifecycle
-(start/complete/error/close), cancellation (client disconnect), event
-serialization, and an `on_finish` persistence hook.
+## 3. Implement callbacks
-Layer 0 is a shared `A2ATransport` with two facades on top:
-`WorkflowEngineClient` (workflow send) and `ExtensionSender` (one-shot
-pre-positioning). See DESIGN.md for the rationale.
+### `on_task(request) -> MessageContent`
----
+The callback receives business data only. Common fields:
-## 3. Quick Integration (Layer 2: `execute_psop`)
+| Field | Meaning |
+|---|---|
+| `execution_id` | local workflow execution identity |
+| `task_id` | stable local logical subtask identity |
+| `input` | current task text or structured data |
+| `agent_name`, `skill` | target business capability |
+| `instruction` | current task description only |
+| `step_name`, `language` | workflow provenance |
+| `workflow_input` | selected upstream history window |
-### 3.1 Minimal Example
+The callback may call the A2A-T SDK and an LLM, then return final A2A parts, metadata, and activated extension URIs:
```python
-import asyncio
-from workflow_engine import (
- execute_psop, ControlPoint, RouteDecision,
- TaskResponse, RegistryClient, load_psop,
-)
+from a2a_t.client import A2ATClient
+from a2a_t.core.standard_templates import PRIVATE_LINE_COMPLAINT_URI
+a2at_client = A2ATClient(env_path=a2at_env_path)
-class MyControlPoint(ControlPoint):
- async def on_task(self, request, engine_client):
- # SDK assembles the full message (context + task + lang hint)
- # in request.message. Just send it.
- result = await engine_client.send_message(
- request.agent_name, request.message
+class Callbacks(ControlPoint):
+ async def on_task(self, request):
+ generated = await asyncio.to_thread(
+ a2at_client.generate_task_prompt_from_text,
+ request.input.text,
+ PRIVATE_LINE_COMPLAINT_URI,
+ )
+ return A2atMessages.from_generated(
+ generated,
+ [Part(text=request.instruction)],
)
- return TaskResponse(success=bool(result.text), output=result.text)
-
- async def on_route(self, step_name, results, conditions):
- return RouteDecision(next_step=conditions[0].step)
-
-
-async def main():
- registry = RegistryClient(url="https://127.0.0.1:5000", ssl_verify=False)
- agent_cards = await registry.fetch_agent_cards()
-
- workflow = await load_psop(
- base_url="https://127.0.0.1:5001",
- psop_id="your-psop-id",
- ssl_verify=False,
- )
-
- # execute_psop builds A2ATransport + WorkflowEngineClient internally
- async for event in execute_psop(
- psop=workflow,
- agent_cards=agent_cards,
- control_point=MyControlPoint(),
- a2at_env_path=".env",
- credentials_config="agent_credentials.json",
- runtime_intent="Diagnose SPN cross-city fault",
- ssl_verify=False,
- ):
- print(f"[{event['type']}] {event['data']}")
-
-
-if __name__ == "__main__":
- asyncio.run(main())
```
-### 3.2 What `execute_psop` Does For You
-
-| Responsibility | Handled by |
-|---------------|-----------|
-| Build A2ATransport, then WorkflowEngineClient over it | `execute_psop` |
-| Attach ControlPoint + event emitter | `execute_psop` |
-| Create WorkflowExecutor with shared emitter | `execute_psop` |
-| Run workflow as asyncio task | `execute_psop` |
-| Emit `start` / `complete` / `error` / `close` lifecycle events | `execute_psop` |
-| Cancel workflow on client disconnect (GeneratorExit) | `execute_psop` |
-| Serialize events to JSON-safe dicts | `execute_psop` |
-| Call `on_finish(result, collected_events)` after workflow ends | `execute_psop` |
-| Call `on_event(event)` transformer per event | `execute_psop` |
-| Close transport httpx pool | `execute_psop` |
-
-### 3.3 Event Types
-
-Events come from three origins. The runner emits the lifecycle bracket
-(`start` ... `complete`/`error` ... `close`); the executor emits step/task
-and routing events; the engine client emits agent traffic and the A2A-T
-extension handlers emit negotiation/authorization/notification.
-
-| Event Type | Origin | When | Data Keys |
-|-----------|-------|------|-----------|
-| `start` | runner | Workflow begins | `workflow`, `steps` |
-| `step_start` | executor | A step begins | `step` |
-| `task_request` | executor | A subtask is dispatched to `on_task` | `step`, `agent`, `task` |
-| `task_response` | executor | `on_task` returned a `TaskResponse` | `step`, `agent`, `task`, `output` |
-| `task_status_changed` | executor | Task status updated | `step`, `subtask_index`, `agent`, `status` |
-| `route_decision` | executor | Branch decision made | `step`, `next`, `reason` |
-| `step_complete` | executor | Step finished | `step`, `results` |
-| `agent_request` | engine client | Message sent to agent | `agent`, `request`, `metadata` |
-| `agent_response` | engine client | Response received | `agent`, `response` |
-| `agent_status_update` | engine client | Streaming status update | `agent`, `state`, `text` |
-| `agent_artifact_update` | engine client | Streaming artifact update | `agent`, `artifact_name`, `text` |
-| `agent_message_event` | engine client | Streaming message event | `agent`, `text` |
-| `negotiation_request` | engine client | Agent needs clarification | `agent`, `round`, `concern` |
-| `negotiation_resolved` | engine client | Clarification provided | `agent`, `round`, `clarification` |
-| `negotiation_failed` | engine client | Negotiation failed | `agent`, `round`, `reason` |
-| `authorization_request` | extension | Agent requests authorization | `agent`, `auth_request` |
-| `authorization_resolved` | extension | Authorization decision | `agent`, `decision` |
-| `notification` | extension | Agent pushes notification | `agent`, `notification` |
-| `workflow_complete` | executor | DAG traversal ended (precedes `complete`/`error`) | (empty) |
-| `complete` | runner | Workflow succeeded | `history`, `step_outputs` |
-| `error` | runner or executor | Workflow failed | runner: `error`, `history`, `step_outputs`; executor: `step`, `results` |
-| `close` | runner | Cleanup done | (empty) |
-
-Compare with constants: `event["type"] == EventType.STEP_START`. Every type
-in the table has a matching constant in `EventType`.
-
-> **`workflow_complete` vs `complete`, and duplicate `error`:** the
-> executor emits `workflow_complete` as soon as DAG traversal ends, then
-> the runner emits `complete` (success) or `error` (failure) with the
-> final `ExecutionResult`. On a step failure the executor emits an
-> `error` carrying `step`/`results`, and the runner later emits a
-> second `error` carrying `history`/`step_outputs` -- check `data` keys
-> to tell them apart.
-
-### 3.4 Persistence Hook (`on_finish`)
+Do not call `WorkflowEngineClient.send_message()` inside this callback. The executor sends the returned content and preserves protocol identity across waiting and negotiation.
-```python
-async def on_finish(result, events):
- """Called after the workflow ends (success or failure)."""
- if result.success:
- save_to_database(result.history, result.step_outputs)
- else:
- log_error(result.error)
-
-
-async for event in execute_psop(..., on_finish=on_finish):
- yield event # SSE stream to client
-```
-
-### 3.5 Event Transformer (`on_event`)
+### `on_self_task(request) -> TaskResult`
-Inject business-specific events or filter SDK events:
+Return zero or more outputs. Outputs may be text or structured JSON values and may contain nested arrays.
```python
-def shape_event(event):
- if event["type"] == "task_status_changed":
- d = event["data"]
- update_my_model(d["step"], d["subtask_index"], d["status"])
- return [
- {"type": "psop_update", "data": {"model": my_model_dump()}},
- event,
- ]
- return event
-
-
-async for event in execute_psop(..., on_event=shape_event):
- ...
+async def on_self_task(self, request):
+ summaries = [summarize(step) for step in request.workflow_input.upstream_results]
+ return TaskResult.succeeded(summaries)
```
-Return values: `event` (pass through), `None` (skip), `list` (inject multiple).
+Use `TaskResult.failed(code, message)` or raise `BusinessFailure` for a safe business failure.
-### 3.6 Cancellation
+### `on_route(request) -> RouteDecision`
-Closing the async iterator (client disconnect, timeout) automatically
-cancels the running workflow. `execute_psop` catches `GeneratorExit`,
-cancels the internal asyncio task, and still calls `on_finish` with a
-"cancelled" result.
+The engine calls once per nonblank edge. `request.current_results` contains results produced by the source step, while `request.workflow_input` contains the selected upstream window.
```python
-async for event in execute_psop(...):
- yield event # StreamingResponse -> client disconnect -> GeneratorExit
+async def on_route(self, request):
+ if evaluate(request.condition, request.current_results):
+ return RouteDecision.allow("condition matched")
+ return RouteDecision.deny("condition did not match")
```
----
-
-## 4. ControlPoint: Flow Decisions
+Unconditional edges never invoke this callback. Multiple conditional edges may be allowed.
-`ControlPoint` drives the workflow forward. You must implement `on_task`
-and `on_route`; the rest have defaults. Authorization-T and Notification-T
-are pre-positioning operations handled via `ExtensionSender` before the
-workflow starts, not in-workflow callbacks.
+### `on_negotiation(request) -> NegotiationReply`
-### 4.1 Methods
+Use the host's A2A-T SDK to process `request.received` and generate a final reply for the same negotiation. `request.previous_exchanges` is ordered history; `remaining_wait_seconds` is the remaining engine budget.
-| Method | Required | Called When | You Decide |
-|--------|----------|-------------|-----------|
-| `on_task(request, engine_client)` | **Yes** | A step sends a task to an agent | whether/how to send, what to return |
-| `on_self_task(request)` | No (default echoes) | A SELF_LOOP step | local result (no A2A-T message) |
-| `on_route(step_name, results, conditions)` | **Yes** | A step has conditional branches | which branch to take |
-| `on_negotiation(agent_name, text, result)` | No (default generic) | Agent returns INPUT_REQUIRED | clarification text |
-
-### 4.2 TaskRequest Fields
+```python
+async def on_negotiation(self, request):
+ from a2a_t.core.standard_templates import (
+ INFORMATION_NEGOTIATION_ACCEPT_REJECT_URI,
+ INFORMATION_NEGOTIATION_PROPOSE_URI,
+ )
+ from a2a_t.negotiation.content import (
+ InformationEndingContent,
+ NegotiationConclusion,
+ NegotiationEndingData,
+ NegotiationItem,
+ )
-| Field | Description |
-|-------|-------------|
-| `agent_name` | Target agent name (matches AgentCard.name) |
-| `skill` | Skill declared on the task |
-| `message` | Full assembled message (upstream context + task + language hint) |
-| `description` | Original task description (for logging/history) |
-| `context` | Upstream context only (without the current task) |
-| `step_name` | Current step name |
-| `subtask_index` | Index within the step's subtasks |
+ context = A2atMessages.negotiation_context(request.received)
+ metadata = request.received.message.metadata
+ remote_prompt = metadata[A2ATExtension.NEGOTIATION_T.uri]
+ validated = a2at_client.validate_propose_prompt_and_data_filling(
+ remote_prompt,
+ context,
+ {
+ "type": "object",
+ "properties": {
+ "items": {
+ "type": "array",
+ "items": {"type": "string"},
+ "minItems": 1,
+ "description": "Business fields requested by the delegated agent",
+ },
+ "relationship": {
+ "type": "string",
+ "nullable": True,
+ "description": "Logical relationship between requested fields",
+ },
+ },
+ "required": ["items"],
+ },
+ INFORMATION_NEGOTIATION_PROPOSE_URI,
+ )
+ supplied = [
+ NegotiationItem(name, resolve_business_value(name, request, validated.data))
+ for name in validated.data["items"]
+ ]
+ generated = a2at_client.generate_negotiation_accept_prompt_from_data(
+ NegotiationEndingData(
+ context,
+ InformationEndingContent(NegotiationConclusion.ACCEPT, supplied),
+ ),
+ INFORMATION_NEGOTIATION_ACCEPT_REJECT_URI,
+ )
+ content = A2atMessages.from_generated(
+ generated,
+ [Part(text="business-supplied clarification")],
+ )
+ return NegotiationReply.send(content)
+```
-### 4.3 Route Decision
+The host validates and interprets the proposal before deciding. Generate Reject or Abort with the corresponding current content API. Accept, Reject, and Abort reuse the received `id`, `round`, and `maxRounds`; the SDK stamps only the terminal `performative`. Do not call the deprecated `start_negotiation`, `receive_negotiation`, or `continue_negotiation` methods, and do not send the legacy `negotiationId/status/extra` shape.
-`conditions` is a `List[JumpCondition]`, each with `.step` (target) and
-`.condition` (description). Return `RouteDecision(next_step=...)`.
+To stop locally without claiming that a protocol Abort was sent:
```python
-async def on_route(self, step_name, results, conditions):
- prompt = build_route_prompt(step_name, results, conditions)
- decision = await my_llm.generate(prompt)
- return RouteDecision(next_step=decision, reason="LLM decided")
+return NegotiationReply.stop("negotiation.rejected", "Host rejected the proposal")
```
-If `next_step` is not in the allowed list, the SDK logs a warning and ends
-the workflow. Empty conditions mean unconditional fan-out (no `on_route`
-call); only conditional branches reach this method.
+If the remote task already exists, a local stop, timeout, cancellation, or invalid negotiation response triggers a bounded best-effort A2A task cancellation. This cleanup does not manufacture an A2A-T Reject or Abort and never replaces the original workflow failure.
-### 4.4 Negotiation
+## 4. Execute
-When an agent returns `INPUT_REQUIRED` and supports Negotiation-T, the
-engine's `send_message` auto-loop calls `on_negotiation` for a clarification
-and resends the follow-up. You only supply the text; do not send messages
-here. Return `None`/`""` to fail the round (the loop stops after
-`max_negotiation_rounds`).
+Low-level execution keeps transport ownership explicit:
```python
-async def on_negotiation(self, agent_name, negotiation_text, receive_result):
- prompt = f"Agent {agent_name} needs: {negotiation_text}"
- return await my_llm.generate(prompt)
+transport = A2ATransport(
+ agent_cards,
+ credentials_config="agent_credentials.json",
+ ssl_verify=True,
+ ca_certs_path="ca.pem",
+ preferred_protocol="HTTP+JSON",
+ send_timeout_seconds=600,
+)
+client = WorkflowEngineClient(transport, max_negotiation_exchanges=3)
+result = await WorkflowExecutor(
+ workflow, Callbacks(), client,
+ runtime_intent="diagnose service",
+ lang="en",
+).run()
+await transport.close()
```
-For reusable strategies, implement `NegotiationStrategy` and inject it into
-`DefaultControlPoint`.
-
----
-
-## 5. Pre-positioning: Authorization-T and Notification-T
-
-Authorization-T and Notification-T are one-shot operations sent via
-`ExtensionSender` **before** the workflow starts. They are not part of the
-in-workflow handler chain (`ExtensionRegistry` only auto-registers Task-T
-and Negotiation-T).
-
-| Operation | Method | Callback mechanism |
-|-----------|--------|-------------------|
-| Authorization-T | `sender.send_authorization(agent, instruction, input)` | Returns `SendMessageResult` directly |
-| Notification-T | `sender.send_notification(agent, instruction, input)` | Returns `SendMessageResult` (subscription confirmed). Subsequent SSE events are handled by the workbench agent's business code, not the SDK |
-
-### Parameters
+The high-level runner yields events and owns only resources that it creates:
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `agent_name` | `str` | Target agent name (must match `AgentCard.name`) |
-| `instruction` | `str` | Short instruction text; becomes `parts[].text` in the A2A message body |
-| `natural_language_input` | `str` | Natural language input passed to the A2A-T SDK to generate a structured extension prompt. The generated value is placed in the message `metadata` under the extension URI key (e.g. `https://.../Authorization-T/v1`). Falls back to this text as-is when SDK generation is unavailable |
+```python
+async for event in execute_psop(
+ workflow,
+ agent_cards,
+ Callbacks(),
+ credentials_config="agent_credentials.json",
+ on_finish=persist,
+):
+ publish(event)
+```
-### Wire format
+If `engine_client` is supplied, the caller must close its transport.
-The resulting A2A message sent to the agent:
+One `WorkflowEngineClient` may be bound to only one active `WorkflowExecutor` at a time. Create separate clients for concurrent workflow executions; transports may be shared only through an adapter that explicitly supports that ownership model.
-```
-parts[].text = instruction
-metadata = { "": "" }
-```
+## 5. Event contract
-Example for Authorization-T:
+Runner lifecycle: `start`, `complete`, `error`, `close`.
-```json
-{
- "parts": [{"text": "Authorize diagnosis operations"}],
- "metadata": {
- "https://projects.tmforum.org/a2aproject/telecommunication/extensions/Authorization-T/v1": ""
- }
-}
-```
+Executor: `step_start`, `step_complete`, `task_request`, `task_response`, `task_status_changed`, `route_decision`, `workflow_complete`.
-### Return value
+Remote interaction: `agent_request`, `agent_response`, `agent_status_update`, `agent_artifact_update`, `agent_message_event`, `negotiation_request`, `negotiation_resolved`, `negotiation_failed`.
-Both methods return `SendMessageResult` containing the agent's response:
+Every executor event contains `execution_id`. A `route_decision` event contains source `step`, target `next`, original `condition`, `conditional`, `allowed`, and `reason`.
-| Field | Description |
-|-------|-------------|
-| `text` | Response text from the agent |
-| `task_state` | Final task state (e.g. `TASK_STATE_COMPLETED`, `WORKING`) |
-| `metadata` | Response metadata (merged task + artifact level) |
+## 6. Authorization-T and Notification-T
-### Example
+Generate and validate content in host code. `A2atMessages.from_generated` keeps the SDK metadata and activates the generated extension, so the target AgentCard must declare that same URI.
```python
-sender = ExtensionSender(transport)
-
-# Authorization-T: send whitelist strategy before workflow
-auth = await sender.send_authorization(
- "SPN Domain Agent",
- "Authorize diagnosis operations",
- "Task type: fault diagnosis, operations: optical module replacement, port reset"
+from a2a_t.core.standard_templates import (
+ AUTHORIZATION_POLICY_MANAGEMENT_URI,
+ SUBSCRIBE_INCIDENT_URI,
)
-# Notification-T: subscribe to recovery results
-notif = await sender.send_notification(
- "SPN Domain Agent",
- "Subscribe to recovery notifications",
- "Topic: service-recovery-execution-result"
+auth_transport = A2ATransport(agent_cards, auth_provider=provider)
+sender = ExtensionSender(auth_transport)
+
+generated_auth = a2at_client.generate_auth_prompt_from_text(
+ authorization_text,
+ AUTHORIZATION_POLICY_MANAGEMENT_URI,
)
+auth_content = A2atMessages.from_generated(generated_auth, [Part(text="authorize policy")])
+auth_result = await sender.send_authorization("domain-a", auth_content)
+if not auth_result.is_success:
+ record_independent_failure(auth_result.failure_code, auth_result.failure_message)
```
----
-
-## 6. Layer 0: A2ATransport + Facades
-
-Use this layer directly when you need manual control of the transport and
-the send lifecycle.
+Notification uses a long-lived handle:
```python
-from workflow_engine import (
- A2ATransport, WorkflowEngineClient, ExtensionSender,
- ControlPoint, WorkflowExecutor,
-)
-
-transport = A2ATransport(
- agent_cards=agent_cards,
- a2at_env_path=".env",
- credentials_config="agent_credentials.json",
- ssl_verify=False,
+def on_notification(subscription, received):
+ process(received)
+ if is_expected_final_result(received):
+ subscription.close()
+
+generated_notification = a2at_client.generate_notification_prompt_from_text(
+ subscription_text,
+ SUBSCRIBE_INCIDENT_URI,
)
-
-# Workflow facade
-engine_client = WorkflowEngineClient(transport)
-
-# One-shot pre-positioning facade (before the workflow)
-sender = ExtensionSender(transport)
-auth_result = await sender.send_authorization("agent_a", "authorize", "Diagnose SPN fault")
-notif_result = await sender.send_notification("agent_a", "subscribe recovery", "Diagnose SPN fault")
-
-# Then run the workflow
-executor = WorkflowExecutor(
- workflow=workflow,
- control_point=MyControlPoint(),
- engine_client=engine_client,
- runtime_intent="Diagnose SPN cross-city fault",
+notification_content = A2atMessages.from_generated(
+ generated_notification,
+ [Part(text="subscribe to business event")],
)
-result = await executor.run()
-await transport.close()
+subscription = sender.open_notification("domain-a", notification_content, on_notification)
+ack = await subscription.acknowledgement
+if ack.is_failure:
+ subscription.close()
+ record_independent_failure(ack.failure_code, ack.failure_message)
+await subscription.completion
```
-Both facades share one transport. `WorkflowEngineClient` owns the workflow
-send path (Task-T generation, Negotiation-T auto-loop, event callback);
-`ExtensionSender` owns one-shot pre-positioning. Neither duplicates wire
-code.
-
----
+Use a transport instance separate from workflow task traffic. Host orchestration may log an independent-operation failure, but it must not make workflow execution depend on that result unless explicit business policy says so.
-## 7. A2A-T Extensions
-
-The four extensions split into in-workflow and one-shot pre-positioning:
-
-| Extension | Lifecycle | Handler | Description |
-|---|---|---|---|
-| Task-T | in-workflow | `TaskTHandler` (auto-registered) | Generates structured task prompt on send |
-| Negotiation-T | in-workflow | `NegotiationTHandler` (auto-registered) | Extracts negotiation context on receive |
-| Authorization-T | pre-positioning | N/A (no handler) | Pre-positioned via `ExtensionSender` |
-| Notification-T | pre-positioning | N/A (no handler) | Subscription via `ExtensionSender` |
-
-The handler chain runs in every `send_message`: `before_send` (Task-T
-injects the prompt), send, `after_receive` (Negotiation-T extracts context,
-feeds the auto-loop). Pre-positioning extensions bypass this chain entirely.
-
-Prompt generation: Task-T uses the A2A-T SDK's `generateTaskPrompt`.
-Authorization-T, Notification-T, and Negotiation-T prompt generators are
-reserved on `ExtensionSender` and wired to the A2A-T SDK as support lands
-upstream; until then the engine falls back to the raw natural-language
-input.
-
----
-
-## 8. Agent Authentication
-
-The SDK provides two authentication layers. They can be used independently or combined (custom provider runs first, credentials-based auth second).
-
-### 8.1 Credentials-Based Auth
-
-When an AgentCard declares `securitySchemes` and `securityRequirements`,
-the SDK logs in to obtain a token and attaches the auth header to outbound
-requests. Configure a JSON credentials file (see README for the full field
-table) and pass the path to `A2ATransport(credentials_config=...)` or
-`execute_psop(..., credentials_config=...)`.
+## 7. Existing task operations
```python
-transport = A2ATransport(
- agent_cards=agent_cards,
- credentials_config="agent_credentials.json",
- ssl_verify=False,
-)
+task = await client.get_task(agent, task_id)
+tasks = await client.list_tasks(agent, ListTasksRequest(page_size=20))
+cancelled = await client.cancel_task(agent, task_id)
+latest = await client.subscribe_to_task(agent, task_id, on_task_event)
```
-A dict may also be passed: `credentials_config={...}`.
-
-**Password Encryption:**
-
-Password fields support the `enc::` format
-(AES-256-GCM). The SDK reads the key from `A2AT_CRED_KEY` (32-byte hex)
-at runtime. To encrypt a password:
+Demo or test cleanup may list and cancel unfinished tasks before a run. This is application behavior; the executor does not cancel unrelated remote tasks automatically.
-```bash
-# Generate key (one-time)
-python -c "import secrets; print(secrets.token_hex(32))"
+## 8. Authentication
-# Encrypt
-export A2AT_CRED_KEY=
-python -c "from workflow_engine.client.credential_crypto import encrypt; print(encrypt('MyPassword'))"
-# Output: enc:xxxxxxxxxxxx:yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy
-```
+Credential configuration is keyed by AgentCard name and security-scheme name. A login response token defaults to the `accessSession` body field and may use a configured nested `token_field`. Passwords may use the engine's encrypted `enc:` form.
-Or programmatically:
+Repeated credentials may be defined once under `profiles` and bound under `agents`; an agent may supply nested `overrides`. Unknown profiles, malformed configuration, missing explicitly configured files, and encrypted values without a valid key fail during transport construction.
-```python
-from workflow_engine.client.credential_crypto import encrypt
-import os
-os.environ["A2AT_CRED_KEY"] = "a1b2c3d4..."
-encrypted = encrypt("MyPassword")
-# encrypted = "enc:xxxxxxxxxxxx:yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy"
+```json
+{
+ "profiles": {
+ "shared-login": {
+ "bearer": {
+ "login_url": "https://identity.example/login",
+ "request_fields": {"username": "user", "password": "enc::"},
+ "token_field": "accessSession"
+ }
+ }
+ },
+ "agents": {
+ "domain-a": {"profile": "shared-login"},
+ "domain-b": {
+ "profile": "shared-login",
+ "overrides": {"bearer": {"request_fields": {"username": "domain-b-user"}}}
+ }
+ }
+}
```
-### 8.2 Custom AuthProvider
-
-For non-standard auth (corporate SSO, external identity providers, agents
-with no `securitySchemes` that still require auth), implement the
-`AuthProvider` ABC:
+Alternatively, supply an `AuthProvider`:
```python
-from workflow_engine import AuthProvider
-
-class SsoAuthProvider(AuthProvider):
- def __init__(self, sso_client):
- self._sso = sso_client
-
- def apply_auth(self, agent_name: str, agent_card, headers: dict) -> None:
- token = self._sso.get_access_token(agent_name)
+class Provider(AuthProvider):
+ def apply_auth(self, agent_name, agent_card, headers):
+ token = token_service.get_or_refresh(agent_name)
headers["Authorization"] = f"Bearer {token}"
```
-Register via `A2ATransport`:
+The provider owns token acquisition. The engine does not read provider-specific usernames, passwords, or login addresses. If configured credentials and the provider produce different values for the same header, the request fails.
-```python
-transport = A2ATransport(
- agent_cards=agent_cards,
- auth_provider=SsoAuthProvider(sso_client),
- ssl_verify=False,
-)
-```
+## 9. TLS and diagnostics
-`apply_auth` is called on **every** message send, regardless of whether
-the AgentCard declares `securitySchemes`. Typical use cases:
+Default TLS verifies the server with system trust. Optional parameters are `ca_certs_path`, `client_cert_path`, `client_key_path`, `client_key_password`, and `crl_path`. Missing or invalid configured files raise during construction.
-- AgentCard has no `securitySchemes` but the server still requires auth
-- Auth tokens come from an external identity provider (e.g. corporate SSO)
-- Non-standard auth headers (e.g. `X-Custom-Token`, `X-Request-Signature`)
+Set `WORKFLOW_ENGINE_PROTOCOL_LOGGING=true` to log pretty-printed final A2A request/response objects. Sensitive headers stay redacted unless the separate sensitive-header flag is explicitly enabled. Protocol logs are diagnostic representations; transport internals may still prevent observation of exact wire bytes.
-### 8.3 Combining Both
+## 10. Workflow retrieval
```python
-transport = A2ATransport(
- agent_cards=agent_cards,
- credentials_config="agent_credentials.json", # credentials auth
- auth_provider=SsoAuthProvider(sso_client), # custom auth (runs first)
- ssl_verify=False,
-)
-```
+matches = await search_psop(base_url, intent, top_n=5, access_token=token)
+workflow = await load_psop(base_url, matches[0].workflow_id, access_token=token)
----
-
-## 9. Integration Checklist
+registry = RegistryClient(registry_url)
+agent_cards = await registry.fetch_agent_cards()
+```
-1. [ ] Install: `pip install workflow-exec-engine`
-2. [ ] Implement `ControlPoint` (at minimum: `on_task` + `on_route`)
-3. [ ] Get AgentCards (from registry or custom source)
-4. [ ] Load a PSOP workflow (`load_psop`) or build a `Workflow` from dict
-5. [ ] Configure agent auth (optional: `credentials_config` or `auth_provider`)
-6. [ ] Pre-position Authorization-T / Notification-T via `ExtensionSender` (optional)
-7. [ ] Run with `execute_psop` (Layer 2) or `WorkflowExecutor` (Layer 1)
-8. [ ] Drain the event stream; persist results in `on_finish`
+`ssl_verify=False` is available for controlled development endpoints with untrusted certificates. It is scoped to the created HTTP client and does not change process-wide TLS defaults.
diff --git a/README.md b/README.md
index 0aad961..2cd1d3e 100644
--- a/README.md
+++ b/README.md
@@ -1,279 +1,127 @@
-# workflow-engine
+# A2A-T Workflow Engine SDK（Python）
-独立工作流执行 SDK，支持宿主 Agent 执行编排中心工作流（PSOP），同时保留对 A2A 通信、A2A-T 扩展与路由决策的完全控制权。SDK 自包含，不依赖编排中心任何代码。
+面向宿主智能体的嵌入式工作流执行引擎。引擎负责工作流 DAG、A2A 消息封装、任务与会话关联、远端任务等待、Negotiation-T 交互循环和生命周期；宿主负责业务输入解释、A2A-T 内容生成与语义校验，以及路由和协商决策。
-> 完整设计参见 [DESIGN.md](DESIGN.md)。本文档面向集成者，快速上手与接口说明。
+Python 与 Java 版本遵循同一业务契约。Python 当前要求 Python 3.12+、`a2a-sdk>=1.1.2,<2` 和 `a2a-t-sdk>=1.0.9,<2`。
-## 设计原则
+## 安装
-| SDK 提供（协议机制） | 用户控制（业务决策） |
-|---|---|
-| A2A 消息发送、流式、SSE 归一化 | 何时 / 是否发送任务 |
-| Agent 认证（Bearer、自定义 Header，基于 AgentCard） | 凭据配置 |
-| A2A-T 扩展（Task-T、Negotiation-T、Authorization-T、Notification-T） | 授权审批、通知处理 |
-| DAG 遍历、上下文组装、状态管理 | 分支路由决策 |
-| 事件追踪 | 事件处理方式 |
-
-## 架构
-
-共享传输层 + 两个门面，职责单一：
-
-```
-A2ATransport（共享通信层：httpx + 认证 + AgentCard 映射 + SSE 消费）
- ├── WorkflowEngineClient（工作流发送门面：Task-T 生成、Negotiation-T 自动循环、事件回调、ControlPoint 装配）
- └── ExtensionSender（一次性预置门面：Authorization-T / Notification-T 发送）
+```bash
+python -m pip install workflow-exec-engine
```
-决策层接口：
-
-- **ControlPoint** — 流程决策（`on_task` / `on_self_task` / `on_route` / `on_negotiation`）
-
-Authorization-T 和 Notification-T 是预置操作，在工作流启动前通过 `ExtensionSender` 单向下发，不在工作流执行中回调。
-
-```mermaid
-flowchart TB
- subgraph User["用户（宿主 Agent）"]
- AC["AgentCards
（注册中心或自定义来源）"]
- CP["ControlPoint
流程决策"]
- end
- subgraph SDK["SDK（自包含）"]
- TR["A2ATransport
共享通信层"]
- WEC["WorkflowEngineClient
工作流发送"]
- ES["ExtensionSender
一次性预置"]
- WE["WorkflowExecutor
DAG 遍历"]
- end
- subgraph Agents["远端 Agents"]
- A1["Agent A"]
- A2["Agent B"]
- end
- AC --> TR
- TR --> WEC
- TR --> ES
- WEC -->|send_message| A1
- WEC -->|send_message| A2
- ES -->|预置发送| A1
- WE -->|on_task/on_route| CP
- WEC -->|on_negotiation| CP
-```
+执行引擎直接使用最新版 A2A-T core 元数据类型识别协议，但不会调用 LLM，也不会替宿主生成或校验 Task-T、Negotiation-T、Authorization-T、Notification-T 的业务内容；这些内容操作仍由宿主调用 `a2a-t-sdk` 完成。
-## 快速开始
+## 最小集成
```python
-import asyncio
+from a2a.types import Part
+from a2a_t.client import A2ATClient
+from a2a_t.core.standard_templates import PRIVATE_LINE_COMPLAINT_URI
from workflow_engine import (
- execute_psop, ControlPoint, RouteDecision,
- TaskResponse, RegistryClient, load_psop,
+ A2ATransport, A2atMessages, ControlPoint, ExtensionSender,
+ MessageContent, NegotiationReply, RouteDecision, TaskResult,
+ WorkflowEngineClient, WorkflowExecutor,
)
-class MyControlPoint(ControlPoint):
- async def on_task(self, request, engine_client):
- # SDK 已组装完整消息（上下文 + 任务 + 语言提示），直接发送
- result = await engine_client.send_message(
- request.agent_name, request.message
+class BusinessCallbacks(ControlPoint):
+ async def on_task(self, request):
+ # a2a-t-sdk 负责内容生成；执行引擎只接收生成结果并发送。
+ generated = a2at.generate_task_prompt_from_text(
+ request.input.text,
+ PRIVATE_LINE_COMPLAINT_URI,
+ )
+ return A2atMessages.from_generated(
+ generated,
+ [Part(text=request.instruction)],
)
- return TaskResponse(success=bool(result.text), output=result.text)
-
- async def on_route(self, step_name, results, conditions):
- # conditions: List[JumpCondition]，每个含 .step 与 .condition
- # 用你的 LLM 或业务逻辑选一个分支
- return RouteDecision(next_step=conditions[0].step)
-
-
-async def main():
- # 1. 获取 AgentCards（注册中心或自定义来源）
- registry = RegistryClient(url="https://127.0.0.1:5000", ssl_verify=False)
- agent_cards = await registry.fetch_agent_cards()
-
- # 2. 加载 PSOP 工作流
- workflow = await load_psop(
- base_url="https://127.0.0.1:5001",
- psop_id="your-psop-id",
- ssl_verify=False,
- )
-
- # 3. 执行：execute_psop 内部构建 A2ATransport + WorkflowEngineClient
- async for event in execute_psop(
- psop=workflow,
- agent_cards=agent_cards,
- control_point=MyControlPoint(),
- a2at_env_path=".env",
- credentials_config="agent_credentials.json",
- runtime_intent="诊断 SPN 跨市故障",
- ssl_verify=False,
- ):
- print(f"[{event['type']}] {event['data']}")
-
-
-if __name__ == "__main__":
- asyncio.run(main())
-```
-
-## 分层入口
-
-| 层 | 入口 | 处理 | 你提供 |
-|---|---|---|---|
-| 2（高） | `execute_psop()` | 事件流、生命周期、取消、onFinish | ControlPoint + AgentCards + 配置 |
-| 1（中） | `WorkflowExecutor` | DAG 遍历、上下文组装、调度 | ControlPoint + WorkflowEngineClient + Workflow |
-| 0（低） | `A2ATransport` + 两个门面 | A2A 发送、认证、扩展、SSE | AgentCards + 配置 |
-
-大多数集成使用第 2 层。需要手动控制时使用第 1 层。仅做一次性预置发送时直接使用 `ExtensionSender`。
-
-## 用户实现的接口
-
-### ControlPoint（流程决策）
-
-| 方法 | 必需 | 调用时机 | 决定 |
-|---|---|---|---|
-| `on_task(request, engine_client)` | 是 | 步骤需向 Agent 发送任务 | 是否 / 如何发送 |
-| `on_self_task(request)` | 否（默认回显） | SELF_LOOP 步骤 | 本地处理结果 |
-| `on_route(step_name, results, conditions)` | 是 | 步骤有条件分支 | 走哪个分支 |
-| `on_negotiation(agent_name, text, result)` | 否（默认通用澄清） | Agent 返回 INPUT_REQUIRED | 补充澄清文本 |
-## A2ATransport + 门面（第 0 层）
+ async def on_self_task(self, request):
+ outputs = aggregate(request.workflow_input.upstream_results)
+ return TaskResult.succeeded(outputs)
-```python
-from workflow_engine import A2ATransport, WorkflowEngineClient, ExtensionSender
+ async def on_route(self, request):
+ # 每条非空条件边独立调用；允许多条边同时返回 true。
+ return RouteDecision.allow() if matches(request.condition) else RouteDecision.deny()
-transport = A2ATransport(
- agent_cards=agent_cards,
- a2at_env_path=".env",
- credentials_config="agent_credentials.json",
- ssl_verify=False,
-)
+ async def on_negotiation(self, request):
+ generated = generate_terminal_negotiation_reply_with_a2at(request)
+ return NegotiationReply.send(
+ A2atMessages.from_generated(generated, [Part(text="补充后的业务内容")])
+ )
-# 工作流发送门面
-engine_client = WorkflowEngineClient(transport)
-# 一次性预置门面（工作流开始前）
-sender = ExtensionSender(transport)
-auth_result = await sender.send_authorization("agent_a", "授权诊断操作", "诊断 SPN 跨市故障")
-notif_result = await sender.send_notification("agent_a", "订阅恢复结果通知", "诊断 SPN 跨市故障")
+a2at = A2ATClient(env_path=a2at_env_path)
+transport = A2ATransport(agent_cards=agent_cards)
+client = WorkflowEngineClient(transport)
+result = await WorkflowExecutor(workflow, BusinessCallbacks(), client).run()
+await transport.close()
```
-两个门面共享同一个 transport，不重复 wire 代码。
-
-**前置操作的回调**：Authorization-T 和 Notification-T 是工作流开始前的一次性操作，通过 `ExtensionSender` 发送。发送结果直接通过返回的 `SendMessageResult` 获取，无需额外的回调接口。
+`on_task` 不接收客户端对象，也不自行发送消息。这样业务代码只负责最终内容，引擎统一保证 A2A 信封、上下文 ID、远端任务 ID、协议头和后续协商仍属于同一次任务交互。
-## A2A-T 扩展
+## 路由规则
-| 扩展 | 归属 | 说明 |
-|---|---|---|
-| Task-T | 工作流链路 | 发送时由 SDK 生成结构化任务提示并注入 `metadata["...Task-T/v1"]` |
-| Negotiation-T | 工作流链路 | 接收时从 `metadata["...NEGOTIATION-T"]` 提取协商上下文，驱动自动循环 |
-| Authorization-T | 一次性预置 | 工作流开始前通过 `ExtensionSender` 发送，`instruction` → `parts[].text`，`natural_language_input` → SDK 生成结构化策略 → `metadata["...Authorization-T/v1"]` |
-| Notification-T | 一次性预置 | 工作流开始前通过 `ExtensionSender` 发送，`instruction` → `parts[].text`，`natural_language_input` → SDK 生成结构化订阅 → `metadata["...Notification-T/v1"]` |
+- 空条件边：无需调用 `on_route`，默认放行。
+- 非空条件边：每条边独立调用一次 `on_route(RouteRequest)`。
+- 一个节点可激活 0 到 N 条后继边；全部拒绝表示该分支正常结束。
+- 同一节点的全部条件判断完成后才调度后继节点；任一判断异常时不调度任何后继节点。
-`ExtensionRegistry` 自动注册 Task-T 与 Negotiation-T（工作流内处理器）；Authorization-T / Notification-T 是预置操作，不自动注册，其 handler 类保留供手动注册处理 Agent 内联推送的数据。
+## 上游结果
-## 智能体认证配置
+引擎不再把前置结果拼成 `Runtime Context` Markdown。回调通过 `request.workflow_input` 获取：
-当 AgentCard 声明 `securitySchemes` 与 `securityRequirements` 时，SDK 自动通过登录接口获取 token 并将认证头附加到出站请求。创建 JSON 文件：
-
-```json
-{
- "agent_a": {
- "bearerAuth": {
- "login_url": "https://127.0.0.1:8080/auth/login",
- "method": "POST",
- "content_type": "application/json",
- "request_fields": { "username": "user", "password": "pass" },
- "token_field": "access_token",
- "token_ttl": 3600,
- "auth_header": "Authorization",
- "auth_header_prefix": "Bearer "
- }
- }
-}
-```
+- `runtime_intent`：本次执行的运行时意图；
+- `upstream_results`：按步骤分组的上游任务结果；
+- 每个任务结果包含有序 `outputs`、来源 Agent/skill、逻辑任务 ID、状态和安全错误信息。
-| 字段 | 必填 | 默认 | 说明 |
-|---|---|---|---|
-| login_url | 是 | - | 获取 token 的 URL |
-| method | 否 | POST | HTTP 方法 |
-| content_type | 否 | application/json | 请求内容类型 |
-| request_fields | 否 | - | 请求体字段（覆盖 username/password） |
-| token_field | 否 | accessSession | 响应中提取 token 的路径（点分隔） |
-| token_ttl | 否 | 3600 | token 缓存时长（秒） |
-| auth_header | 否 | Authorization | 自定义认证头名 |
-| auth_header_prefix | 否 | 空 | token 前缀（如 Bearer） |
-| accept_header | 否 | - | 自定义 Accept 头 |
+工作流的 `context_from` 控制聚合范围：省略时使用直接前驱，`["*"]` 使用全部祖先，空列表表示不传上游结果，指定名称时仅传对应祖先步骤。
-智能体名称须与 AgentCard 的 `name` 一致；认证方案名须与 `securitySchemes` 键一致。也可直接传 dict：`credentials_config=dict`。参见 `examples/agent_credentials.example.json`。
+## 独立协议操作
-**密码加密:**
+Authorization-T 和 Notification-T 不属于工作流因果链，应使用独立 `A2ATransport` 和 `ExtensionSender`：
-`request_fields` 中的密码字段支持 `enc:` 前缀加密格式 `enc::`，算法为 AES-256-GCM。SDK 运行时从 `A2AT_CRED_KEY` 环境变量读取密钥自动解密。
+```python
+sender = ExtensionSender(independent_transport)
+authorization = await sender.send_authorization(agent_name, authorization_content)
+
+subscription = sender.open_notification(agent_name, notification_content, on_notification)
+ack = await subscription.acknowledgement
+# 收到业务结果后由集成方显式关闭：
+subscription.close()
+await subscription.completion
+```
-```bash
-# 1. 生成 32 字节密钥 (仅首次)
-python -c "import secrets; print(secrets.token_hex(32))"
+授权或订阅失败不会自动阻断后续工作流。`send_authorization` 会等待授权任务到达终态，集成方通过 `is_success` 判断结果；订阅确认通过 `is_failure` 排除失败、拒绝或取消。订阅确认与长连接结束是两个不同 Future；`heartbeat` 和 `is_healthy()` 可用于本地存活性判断。
-# 2. 设置密钥环境变量
-export A2AT_CRED_KEY=a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2
+## 远端任务管理
-# 3. 加密密码
-python -c "from workflow_engine.client.credential_crypto import encrypt; print(encrypt('Admin@123'))"
-# 输出: enc:xxxxxxxxxxxx:yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy
+```python
+await client.get_task(agent_name, task_id)
+page = await client.list_tasks(agent_name, list_tasks_request)
+await client.cancel_task(agent_name, task_id)
+await client.subscribe_to_task(agent_name, task_id, on_event)
```
-将输出的 `enc:...` 值填入 credentials JSON 的密码字段即可。
+这些是标准 A2A 任务接口；取消任务不等同于 Negotiation-T Abort。
-### 自定义 AuthProvider
+同一个 `WorkflowEngineClient` 同一时刻只能绑定一个工作流执行；并发执行应创建独立 client。协商在本地停止、超时、取消或协议校验失败时，引擎会对已知远端任务做有界的尽力取消，取消失败只记录日志，不覆盖原始错误。
-对于非标准认证（企业 SSO、外部身份提供商、AgentCard 无 `securitySchemes` 但仍需认证），实现 `AuthProvider` ABC：
+## 认证与 TLS
-```python
-from workflow_engine import AuthProvider
-
-class SsoAuthProvider(AuthProvider):
- def apply_auth(self, agent_name: str, agent_card, headers: dict) -> None:
- token = sso_client.get_access_token(agent_name)
- headers["Authorization"] = f"Bearer {token}"
+`A2ATransport` 支持 AgentCard 声明的凭据配置、自定义 `AuthProvider`、自定义 CA、mTLS 客户端证书、CRL、协议偏好和发送超时。默认验证服务端证书；指定的证书文件缺失或无效时启动失败，不会静默降级为不校验。
-transport = A2ATransport(
- agent_cards=agent_cards,
- auth_provider=SsoAuthProvider(),
-)
-```
+仅受控开发环境可显式配置 `ssl_verify=False`。调用方传入的 `httpx.AsyncClient` 仍由调用方关闭；执行引擎只关闭自己创建的客户端资源。
-两种方式可组合使用：`AuthProvider` 先执行，credentials 认证后执行，各自向请求头注入认证信息。
+## 文档与验证
-## 文件结构
+- [设计说明](DESIGN.md)
+- [集成开发指南](DEVELOPER_GUIDE.md)
+- [英文 README](README_en.md)
+```bash
+python -m pip install -e ".[dev]"
+python -m pytest -q
+python -m build
```
-workflow-exec-engine/
-├── README.md # 本文档
-├── README_en.md # English
-├── DESIGN.md # 设计文档
-├── DEVELOPER_GUIDE.md # 开发者指南
-├── pyproject.toml
-├── examples/
-│ ├── quickstart.py
-│ └── execute_psop_demo.py
-└── workflow_engine/
- ├── __init__.py # 公共 API 导出
- ├── runner.py # execute_psop 高层运行器
- ├── core/ # 核心执行
- │ ├── models.py # 数据模型
- │ ├── context_builder.py
- │ └── executor.py # WorkflowExecutor DAG 遍历
- ├── client/ # 通信层
- │ ├── a2a_transport.py # A2ATransport 共享通信层
- │ ├── engine_client.py # WorkflowEngineClient 工作流门面
- │ ├── extension_sender.py # ExtensionSender 一次性门面
- │ ├── extension_handlers.py
- │ ├── extensions.py # A2ATExtension 枚举
- │ ├── auth_manager.py
- │ ├── credential_service.py
- │ ├── ssl_context.py
- │ └── sse_normalization.py
- ├── control/ # 决策接口
- │ └── control_points.py # ControlPoint + EventType
- └── registry/ # 注册中心集成（可选）
- └── registry_client.py
-```
-
-## 许可证
-
-Apache License 2.0
\ No newline at end of file
diff --git a/README_en.md b/README_en.md
index facc9df..e2bb0f2 100644
--- a/README_en.md
+++ b/README_en.md
@@ -1,279 +1,119 @@
-# workflow-engine
+# A2A-T Workflow Engine SDK for Python
-Standalone workflow execution SDK for A2A-T multi-agent orchestration. The host agent executes orchestration-center workflows (PSOP) while retaining full control over A2A communication, A2A-T extensions, and routing decisions. The SDK is self-contained and does not depend on the orchestration center.
+An embedded workflow execution SDK for host agents. The engine owns workflow DAG execution, A2A envelopes, task/context correlation, remote-task waiting, the Negotiation-T interaction loop, and lifecycle management. The host owns business-input interpretation, A2A-T content generation and semantic validation, routing, and negotiation decisions.
-> For the full design see [DESIGN.md](DESIGN.md). This document targets integrators: quick start and interface reference.
+The Python and Java engines follow the same business contract. This package requires Python 3.12+, `a2a-sdk>=1.1.2,<2`, and `a2a-t-sdk>=1.1.0,<2`.
-## Principle
+## Installation
-| The SDK owns (protocol mechanics) | The host owns (business decisions) |
-|---|---|
-| A2A message send, streaming, SSE normalization | Whether and when to send a task |
-| Agent auth (Bearer, custom headers, from AgentCard) | Credential configuration |
-| A2A-T extensions (Task-T, Negotiation-T, Authorization-T, Notification-T) | Authorization approval, notification handling |
-| DAG traversal, context assembly, state management | Branch routing decisions |
-| Event tracking | Event handling strategy |
-
-## Architecture
-
-A shared transport with two single-responsibility facades:
-
-```
-A2ATransport (shared wire: httpx + auth + agent-card map + SSE consumer)
- |-- WorkflowEngineClient (workflow facade: Task-T generation, Negotiation-T
- | auto-loop, event callback, ControlPoint)
- `-- ExtensionSender (one-shot facade: Authorization-T / Notification-T)
+```bash
+python -m pip install workflow-exec-engine
```
-The decision layer interface:
-
-- **ControlPoint** -- flow decisions (`on_task` / `on_self_task` / `on_route` / `on_negotiation`)
-
-Authorization-T and Notification-T are pre-positioning operations sent via `ExtensionSender` before the workflow starts, not in-workflow callbacks.
-
-```mermaid
-flowchart TB
- subgraph User["User (Host Agent)"]
- AC["AgentCards
(registry or custom)"]
- CP["ControlPoint
flow decisions"]
- end
- subgraph SDK["SDK (self-contained)"]
- TR["A2ATransport
shared wire"]
- WEC["WorkflowEngineClient
workflow send"]
- ES["ExtensionSender
one-shot pre-position"]
- WE["WorkflowExecutor
DAG traversal"]
- end
- subgraph Agents["Remote Agents"]
- A1["Agent A"]
- A2["Agent B"]
- end
- AC --> TR
- TR --> WEC
- TR --> ES
- WEC -->|send_message| A1
- WEC -->|send_message| A2
- ES -->|pre-position send| A1
- WE -->|on_task/on_route| CP
- WEC -->|on_negotiation| CP
-```
+The engine uses the current A2A-T core metadata types for protocol coordination. It does not call an LLM or generate or validate Task-T, Negotiation-T, Authorization-T, or Notification-T business content; the host still invokes `a2a-t-sdk` for those content operations.
-## Quick Start
+## Minimal integration
```python
-import asyncio
+from a2a.types import Part
+from a2a_t.client import A2ATClient
+from a2a_t.core.standard_templates import PRIVATE_LINE_COMPLAINT_URI
from workflow_engine import (
- execute_psop, ControlPoint, RouteDecision,
- TaskResponse, RegistryClient, load_psop,
+ A2ATransport, A2atMessages, ControlPoint, ExtensionSender,
+ MessageContent, NegotiationReply, RouteDecision, TaskResult,
+ WorkflowEngineClient, WorkflowExecutor,
)
-class MyControlPoint(ControlPoint):
- async def on_task(self, request, engine_client):
- # SDK assembles the full message (context + task + lang hint)
- result = await engine_client.send_message(
- request.agent_name, request.message
+class BusinessCallbacks(ControlPoint):
+ async def on_task(self, request):
+ generated = a2at.generate_task_prompt_from_text(
+ request.input.text,
+ PRIVATE_LINE_COMPLAINT_URI,
+ )
+ return A2atMessages.from_generated(
+ generated,
+ [Part(text=request.instruction)],
)
- return TaskResponse(success=bool(result.text), output=result.text)
-
- async def on_route(self, step_name, results, conditions):
- # conditions: List[JumpCondition], each with .step and .condition
- return RouteDecision(next_step=conditions[0].step)
-
-
-async def main():
- # 1. Get AgentCards (registry or custom source)
- registry = RegistryClient(url="https://127.0.0.1:5000", ssl_verify=False)
- agent_cards = await registry.fetch_agent_cards()
-
- # 2. Load a PSOP workflow
- workflow = await load_psop(
- base_url="https://127.0.0.1:5001",
- psop_id="your-psop-id",
- ssl_verify=False,
- )
-
- # 3. Execute: execute_psop builds A2ATransport + WorkflowEngineClient internally
- async for event in execute_psop(
- psop=workflow,
- agent_cards=agent_cards,
- control_point=MyControlPoint(),
- a2at_env_path=".env",
- credentials_config="agent_credentials.json",
- runtime_intent="Diagnose SPN cross-city fault",
- ssl_verify=False,
- ):
- print(f"[{event['type']}] {event['data']}")
-
-
-if __name__ == "__main__":
- asyncio.run(main())
-```
-
-## Layered Entry Points
-
-| Layer | Entry Point | Handles | You Provide |
-|---|---|---|---|
-| 2 (high) | `execute_psop()` | Event stream, lifecycle, cancellation, onFinish | ControlPoint + AgentCards + config |
-| 1 (mid) | `WorkflowExecutor` | DAG traversal, context assembly, dispatch | ControlPoint + WorkflowEngineClient + Workflow |
-| 0 (low) | `A2ATransport` + two facades | A2A send, auth, extensions, SSE | AgentCards + config |
-
-Most integrations use Layer 2. Use Layer 1 for manual control. Use `ExtensionSender` directly for one-shot pre-positioning only.
-
-## Interfaces to Implement
-
-### ControlPoint (flow decisions)
-
-| Method | Required | Called when | Decides |
-|---|---|---|---|
-| `on_task(request, engine_client)` | yes | a step needs to send a task | whether/how to send |
-| `on_self_task(request)` | no (default echoes) | a SELF_LOOP step | local result |
-| `on_route(step_name, results, conditions)` | yes | a step has conditional branches | which branch |
-| `on_negotiation(agent_name, text, result)` | no (default generic) | agent returns INPUT_REQUIRED | clarification text |
-## A2ATransport + Facades (Layer 0)
+ async def on_self_task(self, request):
+ return TaskResult.succeeded(aggregate(request.workflow_input.upstream_results))
-```python
-from workflow_engine import A2ATransport, WorkflowEngineClient, ExtensionSender
+ async def on_route(self, request):
+ return RouteDecision.allow() if matches(request.condition) else RouteDecision.deny()
-transport = A2ATransport(
- agent_cards=agent_cards,
- a2at_env_path=".env",
- credentials_config="agent_credentials.json",
- ssl_verify=False,
-)
+ async def on_negotiation(self, request):
+ generated = generate_terminal_negotiation_reply_with_a2at(request)
+ return NegotiationReply.send(
+ A2atMessages.from_generated(generated, [Part(text="supplemented content")])
+ )
-# Workflow send facade
-engine_client = WorkflowEngineClient(transport)
-# One-shot pre-positioning facade (before workflow starts)
-sender = ExtensionSender(transport)
-auth_result = await sender.send_authorization("agent_a", "authorize diagnosis", "Diagnose SPN cross-city fault")
-notif_result = await sender.send_notification("agent_a", "subscribe to recovery result", "Diagnose SPN cross-city fault")
+a2at = A2ATClient(env_path=a2at_env_path)
+transport = A2ATransport(agent_cards=agent_cards)
+client = WorkflowEngineClient(transport)
+result = await WorkflowExecutor(workflow, BusinessCallbacks(), client).run()
+await transport.close()
```
-Both facades share one transport; no wire code is duplicated.
-
-**Pre-positioning callbacks**: Authorization-T and Notification-T are one-shot operations sent via `ExtensionSender` before the workflow starts. The send result is returned directly as a `SendMessageResult` -- no separate callback interface is needed.
+`on_task` does not receive a transport client and does not send the message. The host prepares final content; the engine keeps the A2A envelope, context ID, remote task ID, headers, waiting, and subsequent negotiation in one interaction.
-## A2A-T Extensions
+## Routing contract
-| Extension | Lifecycle | Description |
-|---|---|---|
-| Task-T | in-workflow | SDK generates a structured task prompt on send and injects into `metadata["...Task-T/v1"]` |
-| Negotiation-T | in-workflow | Extracts negotiation context from `metadata["...NEGOTIATION-T"]` on receive, drives the auto-loop |
-| Authorization-T | one-shot pre-position | Sent via `ExtensionSender` before the workflow. `instruction` → `parts[].text`, `natural_language_input` → SDK generates structured policy → `metadata["...Authorization-T/v1"]` |
-| Notification-T | one-shot pre-position | Sent via `ExtensionSender` before the workflow. `instruction` → `parts[].text`, `natural_language_input` → SDK generates structured subscription → `metadata["...Notification-T/v1"]` |
+- A blank condition is unconditional, bypasses `on_route`, and is allowed.
+- Every nonblank edge gets one independent `on_route(RouteRequest)` call.
+- A node may activate zero through N successors; all denied means a normal branch end.
+- All decisions for the source node finish before successors are scheduled. If one decision fails, no successor of that source is scheduled.
-`ExtensionRegistry` auto-registers Task-T and Negotiation-T (in-workflow handlers); Authorization-T / Notification-T are pre-positioning operations, not auto-registered. Their handler classes are retained for callers that need inline handling of agent-pushed data.
+## Upstream results
-## Agent Authentication
+The engine no longer renders predecessor output into a `Runtime Context` Markdown string. Callbacks receive `request.workflow_input` with the runtime intent and ordered, step-scoped upstream task results. Each task result carries outputs, source agent/skill, logical task ID, status, and safe failure information.
-When an AgentCard declares `securitySchemes` and `securityRequirements`, the SDK logs in to obtain a token and attaches the auth header to outbound requests. Create a JSON file:
-
-```json
-{
- "agent_a": {
- "bearerAuth": {
- "login_url": "https://127.0.0.1:8080/auth/login",
- "method": "POST",
- "content_type": "application/json",
- "request_fields": { "username": "user", "password": "pass" },
- "token_field": "access_token",
- "token_ttl": 3600,
- "auth_header": "Authorization",
- "auth_header_prefix": "Bearer "
- }
- }
-}
-```
+`context_from` selects the history window: omission selects direct predecessors, `["*"]` selects every ancestor, an empty list selects none, and named entries select only those ancestor steps.
-| Field | Required | Default | Description |
-|---|---|---|---|
-| login_url | yes | - | URL to obtain the token |
-| method | no | POST | HTTP method |
-| content_type | no | application/json | request content type |
-| request_fields | no | - | request body fields (overrides username/password) |
-| token_field | no | accessSession | response path to extract token (dot-separated) |
-| token_ttl | no | 3600 | token cache TTL (seconds) |
-| auth_header | no | Authorization | custom auth header name |
-| auth_header_prefix | no | empty | token prefix (e.g. Bearer) |
-| accept_header | no | - | custom Accept header |
+## Independent protocol operations
-Agent names must match the AgentCard `name`; scheme names must match `securitySchemes` keys. A dict may be passed directly: `credentials_config=dict`. See `examples/agent_credentials.example.json`.
+Authorization-T and Notification-T are independent of workflow causality. Give them a separate `A2ATransport` and `ExtensionSender`:
-**Password Encryption:**
+```python
+sender = ExtensionSender(independent_transport)
+authorization = await sender.send_authorization(agent_name, authorization_content)
-Password fields in `request_fields` support the `enc:` prefix format `enc::` using AES-256-GCM. The SDK reads the key from the `A2AT_CRED_KEY` environment variable at runtime to auto-decrypt.
+subscription = sender.open_notification(agent_name, notification_content, on_notification)
+ack = await subscription.acknowledgement
+subscription.close()
+await subscription.completion
+```
-```bash
-# 1. Generate a 32-byte key (one-time)
-python -c "import secrets; print(secrets.token_hex(32))"
+Authorization or subscription failure does not automatically block workflow execution. `send_authorization` waits for a final task state and the caller checks `is_success`; a subscription acknowledgement must be rejected when `is_failure` is true. A subscription acknowledgement and stream completion are separate futures. Use `heartbeat` and `is_healthy()` for local liveness checks.
-# 2. Set the key as an environment variable
-export A2AT_CRED_KEY=a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2
+## Remote task management
-# 3. Encrypt the password
-python -c "from workflow_engine.client.credential_crypto import encrypt; print(encrypt('Admin@123'))"
-# Output: enc:xxxxxxxxxxxx:yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy
+```python
+await client.get_task(agent_name, task_id)
+page = await client.list_tasks(agent_name, list_tasks_request)
+await client.cancel_task(agent_name, task_id)
+await client.subscribe_to_task(agent_name, task_id, on_event)
```
-Paste the `enc:...` output into the password field of your credentials JSON.
+These are standard A2A task operations. Task cancellation is not a Negotiation-T Abort.
-### Custom AuthProvider
+A `WorkflowEngineClient` can be bound to only one active workflow execution at a time; concurrent executions require separate clients. When negotiation ends locally because of stop, timeout, cancellation, or protocol validation failure, the engine makes a bounded best-effort cancellation of a known remote task without replacing the original error.
-For non-standard auth (corporate SSO, external identity providers, agents with no `securitySchemes` that still require auth), implement the `AuthProvider` ABC:
+## Authentication and TLS
-```python
-from workflow_engine import AuthProvider
-
-class SsoAuthProvider(AuthProvider):
- def apply_auth(self, agent_name: str, agent_card, headers: dict) -> None:
- token = sso_client.get_access_token(agent_name)
- headers["Authorization"] = f"Bearer {token}"
+`A2ATransport` supports AgentCard-driven credential configuration, a host `AuthProvider`, custom CA trust, mTLS client identity, CRL checks, protocol preference, and send timeout. Server certificates are verified by default. Missing or invalid configured certificate files fail closed instead of silently disabling verification.
-transport = A2ATransport(
- agent_cards=agent_cards,
- auth_provider=SsoAuthProvider(),
-)
-```
+Only controlled development environments should use `ssl_verify=False`. A caller-supplied `httpx.AsyncClient` remains caller-owned; the engine closes only resources it creates.
-Both approaches can be combined: `AuthProvider` runs first, credentials-based auth runs second, each injecting headers independently.
+## Documentation and verification
-## File Structure
+- [Design](DESIGN.md)
+- [Developer guide](DEVELOPER_GUIDE.md)
+- [Chinese README](README.md)
+```bash
+python -m pip install -e ".[dev]"
+python -m pytest -q
+python -m build
```
-workflow-exec-engine/
-|-- README.md # this document
-|-- README_en.md # English
-|-- DESIGN.md # design document
-|-- DEVELOPER_GUIDE.md # developer guide
-|-- pyproject.toml
-|-- examples/
-| |-- quickstart.py
-| `-- execute_psop_demo.py
-`-- workflow_engine/
- |-- __init__.py # public API exports
- |-- runner.py # execute_psop high-level runner
- |-- core/ # core execution
- | |-- models.py # data models
- | |-- context_builder.py
- | `-- executor.py # WorkflowExecutor DAG traversal
- |-- client/ # communication layer
- | |-- a2a_transport.py # A2ATransport shared wire
- | |-- engine_client.py # WorkflowEngineClient workflow facade
- | |-- extension_sender.py # ExtensionSender one-shot facade
- | |-- extension_handlers.py
- | |-- extensions.py # A2ATExtension enum
- | |-- auth_manager.py
- | |-- credential_service.py
- | |-- ssl_context.py
- | `-- sse_normalization.py
- |-- control/ # decision interfaces
- | `-- control_points.py # ControlPoint + EventType
- `-- registry/ # registry integration (optional)
- `-- registry_client.py
-```
-
-## License
-
-Apache License 2.0
\ No newline at end of file
diff --git a/SECURITY.md b/SECURITY.md
new file mode 100644
index 0000000..ea2833a
--- /dev/null
+++ b/SECURITY.md
@@ -0,0 +1,23 @@
+# Security Policy
+
+## Supported Versions
+
+The latest published release and the current `main` branch receive security
+fixes. Pre-release APIs may change when required to close a vulnerability.
+
+## Reporting a Vulnerability
+
+Do not open a public issue. Email **security@openan.com** with a description,
+reproduction steps, affected versions, and potential impact. Do not include
+real production credentials or customer data.
+
+We aim to acknowledge reports within 48 hours and provide a fix or mitigation
+within 90 days. Coordinated disclosure timing will be agreed with the reporter.
+
+## Security Defaults
+
+- TLS certificate verification is enabled by default.
+- Encrypted credentials fail closed when the key is absent or invalid.
+- Full protocol payload logging is not intended for normal production use.
+- `.env`, credentials, private keys, and generated distributions are excluded
+ from source control.
diff --git a/docs/parity-analysis.md b/docs/parity-analysis.md
deleted file mode 100644
index 4e7fc14..0000000
--- a/docs/parity-analysis.md
+++ /dev/null
@@ -1,354 +0,0 @@
-# Python vs Java A2A-T SDK Parity Analysis
-
-## 1. Overview
-
-### Repositories
-
-| Role | Python | Java |
-|------|--------|------|
-| Execution Engine SDK | `workflow-exec-engine` (`workflow_engine` v0.0.3) | `workflow-exec-engine-java` (`workflow-engine` module) |
-| A2A-T Foundation SDK | bundled in `workflow_engine` (a2a_t pip package) | `a2a-t-sdk-java` (a2a-t-core, a2a-t-client, a2a-t-server, a2a-t-llm, a2a-t-negotiation, a2a-t-prompt, a2a-t-resources) |
-| Demo Agents | `orchestration-center/samples/agents/` | `workflow-exec-engine-java/samples/src/.../agents/` |
-
-### Parity Principle
-
-Both SDKs expose the same business-facing interfaces with the same parameter names, parameter meanings, parameter formats, and event data structures. Language differences (CompletableFuture vs async/await, Lombok vs dataclass) are expected; protocol-level behavior must be identical.
-
-### Verdict: ALIGNED with 6 gaps identified below
-
----
-
-## 2. Control Layer
-
-### 2.1 EventType Constants (22/22 match)
-
-All 22 event type string constants are identical between Python and Java: START, COMPLETE, CLOSE, STEP_START, STEP_COMPLETE, TASK_REQUEST, TASK_RESPONSE, TASK_STATUS_CHANGED, ROUTE_DECISION, WORKFLOW_COMPLETE, AGENT_REQUEST, AGENT_RESPONSE, AGENT_STATUS_UPDATE, AGENT_ARTIFACT_UPDATE, AGENT_MESSAGE_EVENT, NEGOTIATION_REQUEST, NEGOTIATION_RESOLVED, NEGOTIATION_FAILED, AUTHORIZATION_REQUEST, AUTHORIZATION_RESOLVED, NOTIFICATION, ERROR.
-
-### 2.2 EventCallback
-
-| Aspect | Python | Java | Match |
-|--------|--------|------|-------|
-| Method | `on_event(self, event_type: str, data: Dict)` | `onEvent(String eventType, Map data)` | YES |
-| Default behavior | no-op | no-op | YES |
-
-### 2.3 ControlPoint Interface
-
-| Method | Python | Java | Match |
-|--------|--------|------|-------|
-| on_task (required) | `async on_task(request, engine_client) -> TaskResponse` | `CompletableFuture onTask(TaskRequest, WorkflowEngineClient)` | YES |
-| on_self_task (default) | `async on_self_task(request) -> TaskResponse` | `default CompletableFuture onSelfTask(TaskRequest)` | YES |
-| on_route (required) | `async on_route(step_name, results, conditions) -> RouteDecision` | `CompletableFuture onRoute(String, Map, List)` | YES |
-| on_negotiation (default) | `async on_negotiation(agent_name, neg_text, receive_result) -> str` | `default CompletableFuture onNegotiation(String, String, Map)` | YES |
-
-Key: on_self_task does NOT receive engine_client in either SDK. Self-loop tasks must not send A2A-T messages.
-
-### 2.4 DefaultControlPoint + NegotiationStrategy
-
-All 4 method behaviors match: on_task (send + success=text non-empty), on_self_task (echo message), on_route (first non-terminal branch), on_negotiation (delegate to strategy or generic). Constructor accepts NegotiationStrategy in both.
-
----
-
-## 3. Core Models (all fields match)
-
-### 3.1 Data Models
-
-| Model | Fields | Match |
-|-------|--------|-------|
-| Task | agent, skill="", description="", status=PENDING | YES |
-| TaskRequest | agent_name, skill, message, context, step_name, subtask_index=0, description="" | YES |
-| TaskResponse | success, output="", error=None, metadata=None | YES |
-| SendMessageResult | text="", task=None, metadata={}, task_state="" | YES |
-| RouteDecision | next_step, reason="" | YES |
-| JumpCondition | step, condition="" | YES |
-| ExecutionResult | success, history=[], step_outputs={}, error=None | YES |
-| WorkflowSearchResult | workflow_id, workflow_type, name, description, tags=[], created_at, score=1.0, user_intent, related_preflow, tasks_summary | YES |
-
-### 3.2 Enums
-
-| Enum | Values | Match |
-|------|--------|-------|
-| StepType | ALL_SUCCESS="AllSuccess", ANY_SUCCESS="AnySuccess", SELF_LOOP="SelfLoop" | YES |
-| TaskStatus | PENDING="pending", RUNNING="running", SUCCESS="success", FAILED="failed" | YES |
-
-### 3.3 Workflow / WorkflowStep
-
-All fields match: id, name, description, steps, name, subtasks, next, layer, context_from, step_type. from_dict/fromMap parsing: step_type/type fallback, context_from str->list coercion all match.
-
----
-
-## 4. Client Layer
-
-### 4.1 WorkflowEngineClient Interface
-
-| Method | Python | Java | Match |
-|--------|--------|------|-------|
-| send_message | `(agent_name, message, context_id=None, metadata=None, skip_extensions=False)` | `sendMessage(agentName, message, contextId, metadata)` | GAP 2 |
-| set_control_point | `set_control_point(cp)` | `setControlPoint(ControlPoint)` | YES |
-| set_event_callback | `set_event_callback(cb)` | `setEventCallback(EventCallback)` | YES |
-| close | `async close()` | `void close()` | YES |
-
-**GAP 2 (non-critical):** Python has `skip_extensions` param. Not called by executor (executor calls on_self_task directly for SELF_LOOP). No protocol impact.
-
-### 4.2 DefaultWorkflowEngineClient / WorkflowEngineClient (Python)
-
-| Behavior | Match |
-|----------|-------|
-| Constructor: transport + custom_handlers + event_callback + max_negotiation_rounds | YES |
-| ExtensionRegistry: pre-registers Task-T + Negotiation-T | YES |
-| send_message flow: before_send -> AGENT_REQUEST -> transport.send -> after_receive -> auto_negotiate | YES |
-| AGENT_REQUEST data: {agent, request, metadata} | YES |
-| AGENT_RESPONSE data: {agent, response, metadata} | YES |
-| Auto-negotiation: INPUT_REQUIRED -> on_negotiation -> follow-up -> recurse | YES |
-| Follow-up format: [NEGOTIATION_RESOLUTION]\n...clarification...\n---\nOriginal Task:\n...message...\n\nPlease re-execute... | YES |
-| build_negotiation_follow_up_meta: SDK continue_negotiation with AGREED | YES |
-| Fallback: {NEGOTIATION_T.uri: "## Data Return Confirmation\n" + clarification} | YES |
-
-**GAP 3 (resolved):** Python AGENT_RESPONSE includes `metadata`: `{"agent", "response", "metadata"}`. Java now also includes metadata in AGENT_RESPONSE event.
-
-### 4.3 ExtensionHandler + TaskTHandler + NegotiationTHandler
-
-All behaviors match exactly:
-- Skip if no a2at_client
-- Skip if [NEGOTIATION_RESOLUTION] in message (Task-T)
-- Find URI from card capabilities.extensions
-- Skip if metadata already preset (Task-T)
-- Call generate_task_prompt / receive_negotiation
-- On success/failure: same metadata injection + same logging
-- Negotiation: extract context from DATA-NEGOTIATION-T key, fallback to metadata
-- Exception handling: "Unsupported negotiation type" -> debug level
-
-### 4.4 ExtensionInterceptor (A2A-Extensions Header)
-
-Both SDKs filter: only extensions present in message metadata are included in the A2A-Extensions header. This resolves the previous complaint about all extensions being stuffed in.
-
-### 4.5 A2ATExtension Enum (4/4 URIs match)
-
-TASK_T, NEGOTIATION_T, AUTHORIZATION_T, NOTIFICATION_T - all URIs identical.
-
-### 4.6 ExtensionSender
-
-| Method | Match |
-|--------|-------|
-| send_extension_message | YES |
-| send_authorization | YES |
-| send_notification | YES |
-| generate_extension_prompt dispatch | YES |
-
-**GAP 4 (not an SDK gap):** Java's `sendNotification` has a `Consumer