Skip to content

EXTENDING_CONNECTORS.md

Latest commit

 

History

History
397 lines (299 loc) · 15.8 KB

EXTENDING_CONNECTORS.md

File metadata and controls

397 lines (299 loc) · 15.8 KB
Raw

Extending Connectors · 自定义连接器

How to plug your firm's private systems into the runtime by subclassing BaseConnector. Covers the four-method interface, sandbox fixture conventions, testing, and the red lines you cannot cross.

通过继承 BaseConnector 把律所自有系统接入运行时。本文覆盖四方法接口、沙盒 fixture 规范、测试流程、不能踩的红线。

1. The connector contract · 连接器接口

Every connector subclasses connectors/base.py::BaseConnector and implements four methods. The orchestrator does not introspect anything else — keep your subclass focused.

# connectors/base.py
from abc import ABC, abstractmethod
from typing import Any

class BaseConnector(ABC):
    """Every connector must implement these four methods."""

    name: str                                    # e.g., "pkulaw"
    sandbox_fixture_dir: str                     # sandbox/fixtures/connectors/<name>/

    @abstractmethod
    def auth(self) -> None:
        """Establish credentials. Called once per process."""

    @abstractmethod
    def query(self, q: str, **kw) -> dict:
        """Search. Returns a {results: [...], cursor: str|None} envelope."""

    @abstractmethod
    def fetch(self, resource_id: str) -> dict:
        """Pull a single resource by id. Returns the full document."""

    @abstractmethod
    def push(self, payload: dict) -> dict:
        """Write back. Most read-only connectors raise NotImplementedError."""

1.1 Method semantics · 接口语义

auth() -> None

  • Called once when the connector is first loaded.
  • Read credentials from env vars listed in config/connectors.yaml::<name>::*_env. Never read them from disk or hardcode them.
  • Raise ConnectorAuthError on failure. The orchestrator records the failure to the audit log and disables the connector for the rest of the matter.
  • For OAuth flows: cache the token in production/connector-tokens/<name>.json with strict mode 600.
  • For SSO: take a session cookie via the firm's identity provider; refresh on 401.

query(q: str, **kw) -> dict

  • Free-text search OR structured query. The q string is whatever the caller sent; **kw carries filters (date range, jurisdiction, court level, etc.).
  • Must enforce the rate limit declared in config/connectors.yaml::<name>::rate_limit_per_min. Use connectors/base.py::RateLimiter — do not roll your own.
  • Must hash any client name that appears in q before sending to a public endpoint. Use connectors/base.py::hash_client_name. The orchestrator refuses to let a real client name out of the process.
  • Returns:
{
  "results": [
    {"id": "<resource_id>", "title": "...", "summary": "...", "score": 0.92, "meta": {...}},
    ...
  ],
  "cursor": "<opaque>"   # None if no more pages
}

fetch(resource_id: str) -> dict

  • Pulls one resource. Used after query picks a hit.
  • Returns the full document body. For statute / case databases this is the full text; for enterprise-info databases it's the full company record.
  • Must declare which fields can be persisted to the matter folder. Set _persist_safe_fields on the result if the vendor's licence restricts caching.

push(payload: dict) -> dict

  • Write back. Examples: filing an e-sign request, dropping a document into iManage.
  • The connector itself never executes — it stages. The actual "send" requires a partner approval recorded in the audit log. The push method writes the staged artefact and returns a staging_id; the partner runs python orchestrator/runner.py --execute-push <staging_id> to actually transmit.
  • Most public-database connectors should raise NotImplementedError. Read is enough.

2. A complete example · 完整代码示例

Below is a full connector for a fictional firm-internal regulation library called xinglu (兴律). It supports auth (via env-var token), query, fetch, and read-only push (i.e., raises NotImplementedError).

# connectors/databases/xinglu_connector.py
import os
import time
import requests
from connectors.base import BaseConnector, ConnectorAuthError, RateLimiter, hash_client_name


class XingluConnector(BaseConnector):
    """Firm-internal statute & regulation library."""

    name = "xinglu"
    sandbox_fixture_dir = "sandbox/fixtures/connectors/xinglu/"

    def __init__(self, config: dict):
        self.endpoint = config["endpoint"]                # https://internal.firm.cn/xinglu
        self.token_env = config["api_key_env"]            # e.g., XINGLU_API_KEY
        self.rate_limiter = RateLimiter(
            per_minute=config.get("rate_limit_per_min", 60)
        )
        self._token: str | None = None
        self._session = requests.Session()

    def auth(self) -> None:
        token = os.environ.get(self.token_env)
        if not token:
            raise ConnectorAuthError(
                f"{self.name}: env var {self.token_env} is not set"
            )
        # Validate
        r = self._session.get(
            f"{self.endpoint}/v1/whoami",
            headers={"Authorization": f"Bearer {token}"},
            timeout=10,
        )
        if r.status_code != 200:
            raise ConnectorAuthError(f"{self.name}: auth probe failed: {r.text}")
        self._token = token
        self._session.headers.update({"Authorization": f"Bearer {token}"})

    def query(self, q: str, **kw) -> dict:
        self.rate_limiter.wait()
        safe_q = hash_client_name(q)                      # never leak client identifiers
        params = {
            "q": safe_q,
            "limit": kw.get("limit", 20),
            "jurisdiction": kw.get("jurisdiction", "PRC"),
            "category": kw.get("category"),               # statute | judicial-interpretation | regulator-notice
        }
        if "cursor" in kw:
            params["cursor"] = kw["cursor"]
        r = self._session.get(f"{self.endpoint}/v1/search", params=params, timeout=15)
        r.raise_for_status()
        data = r.json()
        return {
            "results": [
                {
                    "id": hit["doc_id"],
                    "title": hit["title"],
                    "summary": hit.get("snippet", ""),
                    "score": hit.get("score", 0.0),
                    "meta": {
                        "jurisdiction": hit.get("jurisdiction"),
                        "effective_date": hit.get("effective_date"),
                        "amended_date": hit.get("amended_date"),
                        "category": hit.get("category"),
                    },
                }
                for hit in data.get("hits", [])
            ],
            "cursor": data.get("next_cursor"),
        }

    def fetch(self, resource_id: str) -> dict:
        self.rate_limiter.wait()
        r = self._session.get(
            f"{self.endpoint}/v1/documents/{resource_id}",
            timeout=15,
        )
        r.raise_for_status()
        doc = r.json()
        doc["_persist_safe_fields"] = [
            "doc_id", "title", "effective_date", "amended_date",
            "jurisdiction", "issuer", "category", "full_text",
        ]
        return doc

    def push(self, payload: dict) -> dict:
        raise NotImplementedError(
            "xinglu is a read-only firm-internal database; no push support."
        )

Register the connector in config/connectors.yaml:

databases:
  xinglu:
    enabled: true
    endpoint: https://internal.firm.cn/xinglu
    api_key_env: XINGLU_API_KEY
    rate_limit_per_min: 60
    adapter: connectors/databases/xinglu_connector.py
    sandbox_fixture_dir: sandbox/fixtures/connectors/xinglu/

And add it to the allow-list:

outbound_web:
  allowlist:
    - internal.firm.cn
    - …(existing entries)

Done. The orchestrator will pick the connector up on next start.

3. Sandbox fixtures · 沙盒数据

For --sandbox mode to work with your new connector, you need fixtures.

3.1 Layout

sandbox/fixtures/connectors/xinglu/
  ├── _index.json            # maps query-hash → fixture filename
  ├── q-<hash>.json          # one file per pre-seeded query
  ├── doc-<id>.json          # one file per fetchable doc
  └── README.md              # what's in here, who maintains it

3.2 Index format

{
  "queries": {
    "f3a89c12": "q-f3a89c12.json",
    "b71d0e44": "q-b71d0e44.json"
  },
  "documents": {
    "PRC-COMPANY-LAW-2023": "doc-prc-company-law-2023.json",
    "PIPL-2021": "doc-pipl-2021.json"
  }
}

The hash is sha256(safe_q)[:8]. The runner computes it from the same hash_client_name helper, so query strings round-trip deterministically.

3.3 Fixture file format

A q-*.json mirrors the query return shape:

{
  "results": [
    {
      "id": "PRC-COMPANY-LAW-2023",
      "title": "中华人民共和国公司法（2023 修订）",
      "summary": "对 2018 修订的全面更新…",
      "score": 0.99,
      "meta": {"jurisdiction": "PRC", "effective_date": "2024-07-01", "category": "statute"}
    }
  ],
  "cursor": null
}

A doc-*.json mirrors the fetch return shape, full text and all. Keep these under 200 KB each — large fixtures should use compressed JSON files.

3.4 Generating fixtures

Use the scaffolding helper:

python scripts/generate-fixture.py \
  --connector xinglu \
  --query "经营者集中申报"

It creates a hashed empty fixture file with placeholders for you to fill (or pre-fill if you have a small known answer set).

4. Testing your connector · 测试流程

Every new connector should pass three layers of tests.

4.1 Unit tests (offline)

tests/connectors/test_<name>.py — instantiate with a mocked HTTP transport, assert each method:

def test_query_envelope(mocked_requests):
    c = XingluConnector(config={"endpoint": "https://x", "api_key_env": "XINGLU_API_KEY"})
    os.environ["XINGLU_API_KEY"] = "test-token"
    c.auth()
    result = c.query("公司法")
    assert "results" in result and "cursor" in result

4.2 Sandbox round-trip

python orchestrator/runner.py --sandbox \
  --workflow draft-nda \
  --matter "smoke-test-xinglu"

The workflow runs entirely against fixtures. If your connector throws FileNotFoundError for the test query, add the missing fixture.

4.3 Staging environment (pre-live)

Once the firm has a staging endpoint for the real vendor, run:

python orchestrator/runner.py --live --connectors xinglu --staging \
  --matter "staging-test-xinglu"

This uses the real auth + real endpoint but writes outputs to production/staging/, not the real matter folder. Run it for a week before flipping to live.

5. Recommended private-system connectors · 推荐对接的私有系统

Most firms eventually want at least the following. Each one has a stub in connectors/ you can extend.

System Type Connector stub Notes
iManage DMS connectors/case-management/imanage_connector.py Read-only first. Use the iManage Work API. SSO via Okta / Azure AD.
NetDocuments DMS connectors/case-management/netdocs_connector.py OAuth 2.0; record token rotation policy.
Aderant Expert / Sierra Billing connectors/billing/aderant_connector.py Read matter + time records; do not write back unless the firm explicitly enables it.
Firm-internal statute DB Database template: connectors/databases/_template.py Most firms have a curated 法规库 / clause library.
Firm-internal contract library Database template: connectors/databases/_template.py Versioned clauses; contract-drafter agent's preferred source.
Firm-internal precedents Database template: connectors/databases/_template.py Partner-curated; redaction rules apply.
Custom CMS DMS connectors/case-management/custom_cms_adapter.py For firms running a self-built case management.

Use the _template.py as a starting point — it has all four methods stubbed with the right error semantics.

6. Firm-side checklist before enabling · 律所对接清单

Before flipping any connector to enabled: true in live mode:

  • Rate limit declared. Set rate_limit_per_min to a value below the vendor's documented ceiling, with headroom.
  • SSO or service account. No shared user passwords. Service accounts scoped read-only where possible.
  • Audit hook on. Every call writes a line to production/audit-log/. Test by opening a synthetic matter and checking three log lines appear.
  • No client data on the public internet. If the connector talks to a non-firm endpoint, confirm the data flow does not include unhashed client names, document classification markers, or personal information. Use the risk-reviewer agent's pre-flight scan.
  • CAC pathway recorded for any non-PRC endpoint. Standard contract / security assessment / certification — one of the three, with proof in config/connectors.yaml.
  • PIPL DPA signed with vendor.
  • Vendor incident notification clause in the contract.
  • Sandbox fixture set exists so engineers can develop without live calls.
  • Disabled by default on first deployment; flip on per-environment.

7. Red lines · 红线

The orchestrator's job is to make these impossible. Your connector must not work around them.

  1. No real client data to non-allow-listed endpoints. Even with a valid CAC posture, the orchestrator checks outbound_web.allowlist first. Adding an endpoint to the list requires a partner signoff hash in production/connector-approvals/.
  2. No silent retries on auth failure. A ConnectorAuthError disables the connector for the matter; the partner sees it.
  3. No persistence of vendor data beyond the matter scope. If the vendor licence prohibits caching, set _persist_safe_fields: [] on every fetch result.
  4. No push without --execute-push <staging_id>. Staging-then-execute is mandatory for any write-back connector.
  5. No bypass of rate limits. Even if the vendor allows bursting, the limiter must hold — over-the-wire bursts are a fast track to a 429 and an audit-log hole.
  6. No use of the audit log for analytics. It is a compliance artefact, not a training set.

8. Worked patterns · 常见模式

8.1 Vendor pagination

Some vendors return cursor-based pagination, others offset-based. Normalise into the cursor field. If the vendor uses offsets, base64-encode the offset into the cursor string.

8.2 Multi-tenant SaaS

Vendor identifies the firm via Tenant-ID header. Read it from config/connectors.yaml::<name>::tenant_env and add to the session headers in auth.

8.3 Streaming responses

For large fetches (e.g., a 100-page judgement), use requests streaming + write to production/matters/<matter-id>/cache/<resource_id>.json. Return the path in fetch rather than the body.

8.4 Webhook back-channels

If the vendor calls the firm back (e-sign completion), do not embed a webhook receiver in the connector. Use a separate webhooks/<vendor>/ handler that writes into production/inbox/ for the orchestrator to pick up on the next phase.

9. Removing a connector · 下线连接器

Sometimes a vendor relationship ends. To retire:

  1. Set enabled: false in config/connectors.yaml.
  2. Trigger the vendor's data-deletion procedure; record the certificate of destruction under production/connector-retirement/<name>/.
  3. Keep the fixtures and the adapter file — past matters in the audit log reference them.
  4. Update the allow-list to drop the endpoint.

Do not delete the connector class file. Audit-log replay needs it to resolve historical calls.

10. See also · 参见

  • docs/ARCHITECTURE.md — where connectors sit in the data flow
  • docs/SANDBOX_MODE.md — fixture conventions
  • docs/COMPLIANCE_DISCLAIMER.md — third-party SaaS posture
  • connectors/base.py — the interface in code
  • connectors/case-management/imanage_connector.py — a worked example for DMS