OpenWorker 深度解析（五）：能力擴充 — Tools、Skills、Personas、MCP 與排程

1"""The catalog is **platform-owned and closed**: third parties get breadth from us adding
2vetted capabilities here and from MCP, never by adding entries. MCP tools are *not* in the
3catalog (see ``PERMISSIONS-AND-INBOX.md``)."""

 1@dataclass
 2class ToolSpec:
 3    name: str
 4    schema: dict[str, Any]      # OpenAI 格式的 function tool schema
 5    func: Callable[..., Any]
 6    metadata: Any = None        # aisuite ToolMetadata or None
 7
 8class ToolRegistry:
 9    def register(self, func, *, metadata=None, schema=None) -> ToolSpec:
10        name = getattr(func, "__name__", None)
11        if not name:
12            raise ValueError("Tool function must have a __name__.")
13        meta = metadata or getattr(func, "__aisuite_tool_metadata__", None)
14        # Allow an explicit schema override (param or a `__coworker_schema__` attribute)
15        # for tools whose signature can't be auto-converted to a valid JSON schema.
16        resolved_schema = (
17            schema or getattr(func, "__coworker_schema__", None) or _schema_for(func)
18        )
19        ...

1"""Permission checks live in the PermissionEngine and are applied by the turn engine,
2not here."""

 1@dataclass(frozen=True)
 2class Capability:
 3    id: str
 4    name: str                    # 人類可讀標籤（安裝同意畫面用）
 5    description: str
 6    build: Callable[[AgentContext], list]
 7    requires: tuple[str, ...] = ()          # "workspace" / "executor" / "todo"
 8    risk: tuple[RiskClass, ...] = (RiskClass.READ,)
 9
10    def available(self, context: AgentContext) -> bool:
11        return all(_REQUIREMENTS[r](context) for r in self.requires)

 1def expand(ids: list[str], context: AgentContext) -> list:
 2    """Expand a persona's ``tools:`` id list into concrete tool callables for this context.
 3    Capabilities whose context prerequisites aren't met are skipped (no shell without an
 4    executor, no files without a workspace)."""
 5    tools: list = []
 6    for cap_id in ids:
 7        cap = capability(cap_id)
 8        if cap.available(context):
 9            tools.extend(cap.build(context))
10    return tools

1def risk_summary(ids: list[str]) -> set[RiskClass]:
2    """The union of risk classes a tool list can produce — for the install-consent screen."""

1def _code_files(context):
2    """Repo-oriented files: single-root, line-numbered/windowed `read_file`. Our `grep` and
3    windowed `read_file` replace aisuite's slower `search_files` / `read_file`/`read_file_lines`."""
4
5def _files(context):
6    """Knowledge-work files: multi-root aware (reads/writes across the session's roots), keeps
7    aisuite's `read_file`/`read_file_lines`. Only our `grep` replaces the slow `search_files`."""

 1---
 2id: ops
 3name: Ops Coworker
 4icon: wrench
 5tagline: Operate and investigate — runbooks, logs, infrastructure
 6family: knowledge
 7tools: [files, search, shell, todo]          # ← 只能引用 catalog 裡的 id
 8messaging: true
 9connectors: true
10recommended_models: [anthropic:claude-opus-4-8, openai:gpt-5.5]
11default_permission_mode: interactive
12description: An operations-focused coworker for investigating incidents…
13recommends:
14  - connector: github
15    reason: confirm deploys and inspect the PRs behind a change
16    tier: core
17  - connector: slack
18    reason: receive alerts and reply to the team in-channel
19    tier: core
20  - mcp: filesystem
21    reason: read runbooks and postmortems from a local folder
22    tier: optional
23---
24You are the Ops Coworker — a careful, methodical operations engineer. …
25（以下是完整的 system prompt）

1"""Format: YAML frontmatter (identity + capability declaration) followed by a markdown body that
2is the system prompt. `persona ⊇ skill` — the same frontmatter-markdown shape as SKILL.md, with
3more structured fields."""

1"""Parsing is strict: an invalid manifest raises ``ManifestError`` rather than silently
2producing a broken persona (a third-party persona must fail loudly)."""
3
4_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
5VALID_FAMILIES = {"code", "knowledge"}
6VALID_WORKSPACES = {"git", "project", "deliverable", "none"}
7VALID_MODES = {"discuss", "plan", "interactive", "custom", "auto"}

1# Persona ids become directory names under the managed install area (and registry keys), so
2# they are restricted to a filesystem-safe slug on every OS: no path separators or `..`
3# (traversal), no `:*?"<>|` (invalid on Windows), bounded length.

1"""A session is born from exactly one persona (recorded as ``SessionRecord.agent``); resolving an
2id always returns its Agent even if the persona was later disabled, so live sessions keep
3working. Disable/surface only affect what the *new-session* picker offers."""

1@dataclass
2class Recommendation:
3    """A connection a persona recommends, surfaced in the per-session connections drawer. ``ref`` is a
4    connector id or an MCP server name; ``reason`` is the value it unlocks; ``tier`` ranks it. Not
5    validated against shipped connectors — a persona may recommend one we don't ship yet."""

1"""Progressive disclosure: at session start only the catalog (name + description) is injected
2into the agent's context; the full body is loaded on demand via the `load_skill` tool."""

1"""Scope = folder location (folder-is-truth): global skills live in ``state_dir()/skills``,
2project skills in ``<workspace>/.coworker/skills``. There is no database; every operation
3is a folder + ``SKILL.md`` operation, which keeps project skills shareable via git for free."""

1"""Disable state is deliberately NOT a marker inside the skill folder: project folders travel
2with the repo and one user's disable must not be committed to teammates. It lives in the
3personal ``state_dir()/skills-settings.json`` instead."""

1skill_loader.rescan()
2allowed = skill_filter() if callable(skill_filter) else skill_filter
3skills_ctx = skill_catalog_text(skill_loader, allowed=allowed)

1# Live skill menu (SKILLS-SPEC §4.1): recomputed every turn like the roots list, so
2# a skill installed/enabled/disabled mid-session applies from the NEXT MESSAGE —
3# no new session, no lost context.

1"""`allowed` gates load_skill: a set is a build-time snapshot; a CALLABLE is consulted
2on every call — the manager passes one so Settings disables apply to live sessions
3immediately, and skills created after the engine was built are still loadable
4(loader rescans on a miss)."""

1def load_skill(name: str) -> dict:
2    skill = loader.get(name)
3    if skill is None:
4        loader.rescan()   # created after this session started? pick it up now
5        skill = loader.get(name)

1"""Build the `save_skill` tool (SKILLS-SPEC §5.2). `requires_approval=True` routes every
2call through the standard approval card — the tool's ARGUMENTS are the review surface,
3which is why the schema carries the full instructions and file list. Bundled files may
4only be read from `allowed_dirs` (the session's roots): the worker must never bundle
5arbitrary machine paths into a skill."""

1"""Memory tools — the agent's explicit write paths into memory.
2
3`remember` saves a new fact; `memory_update` / `memory_forget` revise or retire one by
4the [#id] shown in the known-memories block, so corrections replace stale facts instead
5of piling up next to them."""

1def format_memories(items: list[MemoryItem]) -> str:
2    """Render memories for injection into the system prompt. Ids are shown so the agent
3    can revise a memory (`memory_update`) or retire it (`memory_forget`)."""
4    lines = [f"- [#{item.id}] {item.content}" for item in items]
5    return "Known memories (from earlier sessions):\n" + "\n".join(lines)

 1_MEMORY_GUIDANCE = """\
 2Memory:
 3- You have persistent memory across sessions. Use `remember` for durable facts: the user's \
 4corrections and stated preferences (include the why), and project context you couldn't \
 5rederive from the code. Don't save what the repo already records (code structure, git \
 6history, AGENTS.md) or details that only matter to the current task. Use absolute dates, \
 7never "yesterday".
 8- Before saving, check the known-memories list: if an entry already covers it, revise that \
 9entry with `memory_update` instead of adding a near-duplicate; retire wrong or obsolete \
10entries with `memory_forget`.
11- Memories reflect when they were written. If one names a file, flag, or URL, verify it \
12still exists before relying on it."""

1# When-to-remember rules, injected only when a memory store is wired. Without these,
2# models either never call `remember` or save noise the repo already records.

1"""MCPManager — our own thin async MCP client over the official `mcp` SDK.
2
3Async-native (no `nest_asyncio`, no second event loop): each server runs in a dedicated
4asyncio task that opens the transport + `ClientSession`, keeps them alive until shutdown,
5then closes them in the *same* task — required because the SDK's transports use anyio cancel
6scopes that must be entered and exited on one task. Tool calls are awaited from any task on
7the same loop, which is safe."""

 1async def _serve(self, server, ready: asyncio.Future, *, interactive=False) -> None:
 2    try:
 3        async with AsyncExitStack() as stack:
 4            ...
 5            session = await stack.enter_async_context(ClientSession(read, write))
 6            await session.initialize()
 7            listed = await session.list_tools()
 8            conn = _Conn(session, list(listed.tools))
 9            if not ready.done():
10                ready.set_result(conn)      # ← 用 Future 把連線交給呼叫者
11            await conn.shutdown.wait()      # ← task 停在這裡，直到關閉訊號
12    except Exception as exc:
13        if not ready.done():
14            ready.set_exception(exc)        # ← 連線失敗也要傳播出去
15    finally:
16        self._conns.pop(server.name, None)
17        self._tasks.pop(server.name, None)

1def _invoke(_remote: str = remote, **kwargs: Any) -> Any:
2    future = asyncio.run_coroutine_threadsafe(call_async(_remote, kwargs), loop)
3    return future.result(timeout)

1_NAME_OK = re.compile(r"[^a-zA-Z0-9_-]")
2_MAX_NAME = 64  # OpenAI function-name limit
3
4def tool_name(server: str, tool: str) -> str:
5    """`mcp__<server>__<tool>`, sanitized to OpenAI's `[A-Za-z0-9_-]{1,64}` rule."""
6    base = f"mcp__{_NAME_OK.sub('_', server)}__{_NAME_OK.sub('_', tool)}"
7    if len(base) > _MAX_NAME:
8        base = base[:_MAX_NAME]
9    return base

1# We attach the schema + metadata explicitly (rather than via `ai.tool`, which would
2# try to derive a schema from this `**kwargs` wrapper): the registry reads both attrs.
3_invoke.__aisuite_tool_metadata__ = ai.ToolMetadata(
4    name=name, category="mcp", risk_level="medium",
5    capabilities=[server.name], requires_approval=server.requires_approval,
6)
7_invoke.__coworker_schema__ = _openai_schema(name, mcp_tool)

1async def ensure(self, server, *, interactive: bool = False) -> _Conn:
2    """`interactive=True` (explicit connect actions only) lets an OAuth server run
3    the browser sign-in flow; the default refuses it — stored tokens and silent
4    refresh still work, but a server that insists on re-authorization raises
5    InteractiveAuthRequired instead of hijacking the user's browser."""

1"""These tools are intentionally local-first: credentials are read from the SecretStore at
2execution time and never enter prompts. OAuth-managed setup can later replace the manual
3access-token fields without changing the tool surface."""

1def _schema(name, description, properties, required) -> dict:
2    return {"type": "function", "function": {...}}

1def _meta(name, *, approval: bool = False, capabilities=None):
2    return ai.ToolMetadata(
3        name=name,
4        category="connector",
5        risk_level="medium" if approval else "low",
6        capabilities=capabilities or ["integration"],
7        requires_approval=approval,
8    )

 1def _enabled_connector_tools(secrets: SecretStore) -> tuple[set[str], set[str]]:
 2    connectors = {c["name"]: c for c in connector_list(secrets)}
 3    enabled_connectors = {
 4        name for name, c in connectors.items()
 5        if c.get("connected") and c.get("enabled")
 6    }
 7    enabled_tools = {
 8        tool["name"]
 9        for c in connectors.values()
10        if c.get("name") in enabled_connectors
11        for tool in c.get("tools", [])
12        if tool.get("enabled")
13    }
14    return enabled_connectors, enabled_tools

 1"""The `explore` tool — a read-only research subagent with its own context window.
 2
 3Broad questions ("where is retry logic handled?") burn the main session's context on
 4dozens of file reads. `explore` spawns a child TurnEngine over the same workspace with
 5read-only tools and a fresh context; only its final report returns to the caller.
 6
 7The child runs in plan mode — the PermissionEngine hard-blocks writes/shell no matter
 8what the child decides — with no approver, so it never needs an approval round-trip.
 9That's what lets `explore` carry low-risk metadata, which in turn makes several explores
10in one assistant turn eligible for the engine's parallel execution. No recursion: the
11child registry has no `explore` tool."""

1permissions = PermissionEngine(workspace_root=Path(ws), mode=Mode.PLAN)   # ① 硬性唯讀
2_CHILD_MAX_ITERATIONS = 10                                                # ② 迭代上限
3# 子代理的 registry 只註冊 file/git/search 工具 —— 沒有 explore              # ③ 不遞迴

1EXPLORER_INSTRUCTIONS = """You are a read-only code explorer working inside the user's workspace. \
2Answer the research task you're given by searching and reading the code (`grep`, `read_file`, \
3`list_files`, `git_log`, `git_status`, `git_diff`). You cannot write files or run commands.
4
5Your final message is your report — it goes back to the agent that spawned you, not to the \
6user. Make it self-contained: answer the task directly, reference code as path:line, quote the \
7key snippets, and note anything surprising you found along the way. If you couldn't find \
8something, say what you searched so the caller doesn't repeat the same searches."""

1# Tools execute in a worker thread (no running loop), so asyncio.run is safe.
2report, status = asyncio.run(_run())

1"""The scheduler loop — runs in the always-on server.
2
3Policy (agreed): **run-once-catch-up** for runs missed while down (due tasks fire once on
4startup, then resume), and **skip-on-overlap** (don't stack a run if the previous is still
5going). The actual execution is injected as `runner(task, trigger) -> TaskRun` so this stays
6independent of the engine/manager."""

1async def _tick(self, *, trigger: str) -> None:
2    for task in self.store.due():
3        # Spawn, don't await: a run can suspend on a parked approval (standing
4        # scoped approvals, §25) and one blocked automation must never stall the
5        # scheduler loop, other due tasks, or self-wake resumption. Overlap is
6        # still guarded inside run_task via _running_ids.
7        spawned = asyncio.create_task(self.run_task(task, trigger=trigger))
8        self._spawned.add(spawned)
9        spawned.add_done_callback(self._spawned.discard)

1"""Self-wake — tools that let a long-running agent suspend and be re-invoked on a trigger.
2
3Converts an always-on agent into suspend/resume (event-driven, ~zero idle cost): the session
4sleeps and the runtime re-invokes it when a wake is due. Two triggers here: a **timer**
5(`sleep_for` / `sleep_until`) and **on-completion** (`wake_on` a backgrounded job)."""

1if self.extra_tick is not None:
2    try:
3        await self.extra_tick()
4    except Exception:
5        logger.exception("scheduler extra_tick (wake resume) failed")

1def grant_entries(permissions: Any) -> list[str]:
2    """Validate a proposed `permissions` list (from the create-tool schema or the GUI
3    create payload) down to the entries actually grantable. Only `access: "write"` items
4    become grants; the tool must declare a target argument (which excludes exec/destructive
5    tools by construction) and the target must be non-empty. Reads are disclosure-only —
6    rendered on the consent card, never stored. Anything else is dropped, fail-closed."""