Skip to content

Foundational Skills — API

The five foundational skills (discovery, execution, auth_gates, browser, pdf_extraction) provide the cross-cutting capabilities every agent in the catalog can compose with. Decision rationale lives in the ADR index; operational guidance lives in the skill pages.

Skill Discovery

mirai_shared_skills.discovery

Skill Discovery foundational module.

__all__ module-attribute

__all__ = ['SkillDiscoverySkill']

SkillDiscoverySkill

Bases: BaseSkill

Dynamic discovery surface for the shared skill catalog.

description property

description: str

instructions property

instructions: str

name property

name: str

get_tools

get_tools() -> list[Tool]
Source code in mirai_shared_skills/discovery/skill.py
def get_tools(self) -> list[Tool]:
    async def list_skills() -> str:
        """Return every registered skill descriptor as JSON."""
        payload = [_serialize(d) for d in all_descriptors().values()]
        return json.dumps({"skills": payload}, indent=2)

    async def search_skills(query: str) -> str:
        """Return descriptors whose name or description matches the query."""
        payload = [_serialize(d) for d in find(query)]
        return json.dumps({"query": query, "matches": payload}, indent=2)

    async def load_skill_instructions(skill_name: str) -> str:
        """Return the step-by-step instructions for the named skill."""
        try:
            descriptor = get(skill_name)
        except KeyError:
            return json.dumps(
                {"error": "unknown_skill", "skill_name": skill_name},
                indent=2,
            )
        return json.dumps(
            {
                "name": descriptor.name,
                "instructions": descriptor.instructions,
                "tools": list(descriptor.tools),
                "references": list(descriptor.references),
            },
            indent=2,
        )

    return [
        tool_from_function(async_fn=list_skills),
        tool_from_function(async_fn=search_skills),
        tool_from_function(async_fn=load_skill_instructions),
    ]

skill

find-skills — dynamic catalog discovery foundational skill.

Allows an autonomous agent to query the catalog metadata registry, retrieve instructional payloads for individual skills, and bootstrap further reasoning without hard-coded knowledge of available capabilities.

SkillDiscoverySkill

Bases: BaseSkill

Dynamic discovery surface for the shared skill catalog.

description property
description: str
instructions property
instructions: str
name property
name: str
get_tools
get_tools() -> list[Tool]
Source code in mirai_shared_skills/discovery/skill.py
def get_tools(self) -> list[Tool]:
    async def list_skills() -> str:
        """Return every registered skill descriptor as JSON."""
        payload = [_serialize(d) for d in all_descriptors().values()]
        return json.dumps({"skills": payload}, indent=2)

    async def search_skills(query: str) -> str:
        """Return descriptors whose name or description matches the query."""
        payload = [_serialize(d) for d in find(query)]
        return json.dumps({"query": query, "matches": payload}, indent=2)

    async def load_skill_instructions(skill_name: str) -> str:
        """Return the step-by-step instructions for the named skill."""
        try:
            descriptor = get(skill_name)
        except KeyError:
            return json.dumps(
                {"error": "unknown_skill", "skill_name": skill_name},
                indent=2,
            )
        return json.dumps(
            {
                "name": descriptor.name,
                "instructions": descriptor.instructions,
                "tools": list(descriptor.tools),
                "references": list(descriptor.references),
            },
            indent=2,
        )

    return [
        tool_from_function(async_fn=list_skills),
        tool_from_function(async_fn=search_skills),
        tool_from_function(async_fn=load_skill_instructions),
    ]

Execution & Debugging

mirai_shared_skills.execution

Execution & Debugging foundational module.

__all__ module-attribute

__all__ = ['ExecutionDebuggingSkill']

ExecutionDebuggingSkill

Bases: BaseSkill

Raw skill that executes commands in an isolated working directory and reports the captured streams so the agent can verify success objectively.

description property

description: str

instructions property

instructions: str

name property

name: str

get_tools

get_tools() -> list[Tool]
Source code in mirai_shared_skills/execution/skill.py
def get_tools(self) -> list[Tool]:
    async def run_command(
        command: str,
        timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
    ) -> str:
        """Run `command` in a fresh temporary directory and return JSON-encoded
        STDOUT/STDERR plus the exit code so the agent can verify success."""
        with tempfile.TemporaryDirectory(prefix="mirai-exec-") as sandbox:
            returncode, stdout, stderr, timed_out = await _run_subprocess(
                command,
                cwd=sandbox,
                timeout_seconds=timeout_seconds,
            )
        return json.dumps(
            {
                "command": command,
                "returncode": returncode,
                "stdout": stdout,
                "stderr": stderr,
                "timed_out": timed_out,
            },
            indent=2,
        )

    async def read_log_file(path: str, max_bytes: int = MAX_LOG_BYTES) -> str:
        """Tail the given log file and return its contents truncated to
        `max_bytes` so the agent can read residual evidence after a run."""
        target = Path(path).expanduser()
        if not target.is_file():
            return json.dumps({"error": "not_a_file", "path": str(target)}, indent=2)
        data = target.read_bytes()
        truncated = len(data) > max_bytes
        content = data[-max_bytes:].decode("utf-8", errors="replace")
        return json.dumps(
            {
                "path": str(target),
                "truncated": truncated,
                "size_bytes": len(data),
                "content": content,
            },
            indent=2,
        )

    return [
        tool_from_function(async_fn=run_command),
        tool_from_function(async_fn=read_log_file),
    ]

skill

Execution & Debugging — Raw skill that runs commands in an isolated sandbox.

This skill forces the agent to verify code success objectively by reading the captured STDOUT/STDERR plus any rotated log file, drastically reducing silent failures. It is categorised as Raw because arbitrary subprocess execution and filesystem reads are state-mutating relative to the host machine. Downstream clients are expected to wrap every tool exposed here with a SecureSkill policy.

DEFAULT_TIMEOUT_SECONDS module-attribute

DEFAULT_TIMEOUT_SECONDS = 30

MAX_LOG_BYTES module-attribute

MAX_LOG_BYTES = 256 * 1024

MAX_OUTPUT_BYTES module-attribute

MAX_OUTPUT_BYTES = 64 * 1024

ExecutionDebuggingSkill

Bases: BaseSkill

Raw skill that executes commands in an isolated working directory and reports the captured streams so the agent can verify success objectively.

description property
description: str
instructions property
instructions: str
name property
name: str
get_tools
get_tools() -> list[Tool]
Source code in mirai_shared_skills/execution/skill.py
def get_tools(self) -> list[Tool]:
    async def run_command(
        command: str,
        timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
    ) -> str:
        """Run `command` in a fresh temporary directory and return JSON-encoded
        STDOUT/STDERR plus the exit code so the agent can verify success."""
        with tempfile.TemporaryDirectory(prefix="mirai-exec-") as sandbox:
            returncode, stdout, stderr, timed_out = await _run_subprocess(
                command,
                cwd=sandbox,
                timeout_seconds=timeout_seconds,
            )
        return json.dumps(
            {
                "command": command,
                "returncode": returncode,
                "stdout": stdout,
                "stderr": stderr,
                "timed_out": timed_out,
            },
            indent=2,
        )

    async def read_log_file(path: str, max_bytes: int = MAX_LOG_BYTES) -> str:
        """Tail the given log file and return its contents truncated to
        `max_bytes` so the agent can read residual evidence after a run."""
        target = Path(path).expanduser()
        if not target.is_file():
            return json.dumps({"error": "not_a_file", "path": str(target)}, indent=2)
        data = target.read_bytes()
        truncated = len(data) > max_bytes
        content = data[-max_bytes:].decode("utf-8", errors="replace")
        return json.dumps(
            {
                "path": str(target),
                "truncated": truncated,
                "size_bytes": len(data),
                "content": content,
            },
            indent=2,
        )

    return [
        tool_from_function(async_fn=run_command),
        tool_from_function(async_fn=read_log_file),
    ]

Authentication Gates

mirai_shared_skills.auth_gates

Authentication Gates foundational module.

__all__ module-attribute

__all__ = ['AuthenticationGatesSkill', 'CredentialHandoff']

AuthenticationGatesSkill

Bases: BaseSkill

Standard skill that intercepts 401/403 responses and emits handoff signals.

description property

description: str

instructions property

instructions: str

name property

name: str

get_tools

get_tools() -> list[Tool]
Source code in mirai_shared_skills/auth_gates/skill.py
def get_tools(self) -> list[Tool]:
    async def inspect_status_code(status_code: int) -> str:
        """Classify a status code and report whether authentication failed."""
        failed = status_code in AUTH_FAILURE_STATUSES
        return json.dumps(
            {
                "status_code": status_code,
                "authentication_failed": failed,
                "action": "halt_and_handoff" if failed else "continue",
            },
            indent=2,
        )

    async def request_credential_handoff(
        integration: str,
        credential_kind: str,
        status_code: int,
        reason: str,
        docs_url: str | None = None,
    ) -> str:
        """Suspend operations and emit the structured handoff payload."""
        handoff = CredentialHandoff(
            integration=integration,
            credential_kind=credential_kind,
            status_code=status_code,
            reason=reason,
            docs_url=docs_url,
        )
        return json.dumps(handoff.to_payload(), indent=2)

    return [
        tool_from_function(async_fn=inspect_status_code),
        tool_from_function(async_fn=request_credential_handoff),
    ]

CredentialHandoff dataclass

CredentialHandoff(
    integration: str,
    credential_kind: str,
    status_code: int,
    reason: str,
    docs_url: str | None = None,
)

Structured payload describing a credential the human operator must provide.

credential_kind instance-attribute

credential_kind: str

docs_url class-attribute instance-attribute

docs_url: str | None = None

integration instance-attribute

integration: str

reason instance-attribute

reason: str

status_code instance-attribute

status_code: int

to_payload

to_payload() -> dict[str, object]
Source code in mirai_shared_skills/auth_gates/skill.py
def to_payload(self) -> dict[str, object]:
    payload: dict[str, object] = {
        "type": "credential_handoff",
        "integration": self.integration,
        "credential_kind": self.credential_kind,
        "status_code": self.status_code,
        "reason": self.reason,
    }
    if self.docs_url is not None:
        payload["docs_url"] = self.docs_url
    return payload

skill

Authentication Gates — intercept HTTP 401/403 responses and hand off to a human.

The skill exposes a deterministic interceptor that inspects an HTTP status code, suspends the autonomous loop when the upstream rejected the request, and emits a structured handoff payload describing which credential is missing.

Categorised as Standard because the interceptor itself never mutates external state; it only emits structured signals back to the orchestrator.

AUTH_FAILURE_STATUSES module-attribute

AUTH_FAILURE_STATUSES: Final[frozenset[int]] = frozenset(
    {401, 403}
)

AuthenticationGatesSkill

Bases: BaseSkill

Standard skill that intercepts 401/403 responses and emits handoff signals.

description property
description: str
instructions property
instructions: str
name property
name: str
get_tools
get_tools() -> list[Tool]
Source code in mirai_shared_skills/auth_gates/skill.py
def get_tools(self) -> list[Tool]:
    async def inspect_status_code(status_code: int) -> str:
        """Classify a status code and report whether authentication failed."""
        failed = status_code in AUTH_FAILURE_STATUSES
        return json.dumps(
            {
                "status_code": status_code,
                "authentication_failed": failed,
                "action": "halt_and_handoff" if failed else "continue",
            },
            indent=2,
        )

    async def request_credential_handoff(
        integration: str,
        credential_kind: str,
        status_code: int,
        reason: str,
        docs_url: str | None = None,
    ) -> str:
        """Suspend operations and emit the structured handoff payload."""
        handoff = CredentialHandoff(
            integration=integration,
            credential_kind=credential_kind,
            status_code=status_code,
            reason=reason,
            docs_url=docs_url,
        )
        return json.dumps(handoff.to_payload(), indent=2)

    return [
        tool_from_function(async_fn=inspect_status_code),
        tool_from_function(async_fn=request_credential_handoff),
    ]

CredentialHandoff dataclass

CredentialHandoff(
    integration: str,
    credential_kind: str,
    status_code: int,
    reason: str,
    docs_url: str | None = None,
)

Structured payload describing a credential the human operator must provide.

credential_kind instance-attribute
credential_kind: str
docs_url class-attribute instance-attribute
docs_url: str | None = None
integration instance-attribute
integration: str
reason instance-attribute
reason: str
status_code instance-attribute
status_code: int
to_payload
to_payload() -> dict[str, object]
Source code in mirai_shared_skills/auth_gates/skill.py
def to_payload(self) -> dict[str, object]:
    payload: dict[str, object] = {
        "type": "credential_handoff",
        "integration": self.integration,
        "credential_kind": self.credential_kind,
        "status_code": self.status_code,
        "reason": self.reason,
    }
    if self.docs_url is not None:
        payload["docs_url"] = self.docs_url
    return payload

Agent Browser

mirai_shared_skills.browser

Agent Browser foundational module.

__all__ module-attribute

__all__ = ['AgentBrowserSkill']

AgentBrowserSkill

AgentBrowserSkill(fetcher: Fetcher | None = None)

Bases: BaseSkill

Standard skill that performs headless GET requests against the public web.

Source code in mirai_shared_skills/browser/skill.py
def __init__(self, fetcher: Fetcher | None = None) -> None:
    self._fetcher: Fetcher = fetcher or _httpx_fetcher

description property

description: str

instructions property

instructions: str

name property

name: str

get_tools

get_tools() -> list[Tool]
Source code in mirai_shared_skills/browser/skill.py
def get_tools(self) -> list[Tool]:
    async def browse_url(
        url: str,
        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
        user_agent: str = DEFAULT_USER_AGENT,
    ) -> str:
        """Fetch `url` and return JSON with status, final URL, and body text."""
        try:
            response = await self._fetcher(url, timeout_seconds, user_agent)
        except httpx.HTTPError as exc:
            return json.dumps(
                {"error": "http_error", "url": url, "detail": str(exc)},
                indent=2,
            )
        text = html_to_text(response.text)[:MAX_BODY_CHARS]
        return json.dumps(
            {
                "url": url,
                "final_url": str(response.url),
                "status_code": response.status_code,
                "content_type": response.headers.get("content-type"),
                "text": text,
            },
            indent=2,
        )

    return [tool_from_function(async_fn=browse_url)]

skill

Agent Browser — headless web navigation for real-time external intelligence.

The skill uses httpx.AsyncClient for transport and a small stdlib HTML-to-text extractor so the package has no heavyweight runtime dependencies. Downstream clients can replace the Fetcher callable with a Playwright-backed one when JavaScript rendering is required.

DEFAULT_TIMEOUT_SECONDS module-attribute

DEFAULT_TIMEOUT_SECONDS: Final[float] = 15.0

DEFAULT_USER_AGENT module-attribute

DEFAULT_USER_AGENT: Final[str] = (
    "MiraiSharedSkills/0.1 (+https://github.com/mirai-srl)"
)

Fetcher module-attribute

Fetcher = Callable[[str, float, str], Awaitable[Response]]

MAX_BODY_CHARS module-attribute

MAX_BODY_CHARS: Final[int] = 32000

AgentBrowserSkill

AgentBrowserSkill(fetcher: Fetcher | None = None)

Bases: BaseSkill

Standard skill that performs headless GET requests against the public web.

Source code in mirai_shared_skills/browser/skill.py
def __init__(self, fetcher: Fetcher | None = None) -> None:
    self._fetcher: Fetcher = fetcher or _httpx_fetcher
description property
description: str
instructions property
instructions: str
name property
name: str
get_tools
get_tools() -> list[Tool]
Source code in mirai_shared_skills/browser/skill.py
def get_tools(self) -> list[Tool]:
    async def browse_url(
        url: str,
        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
        user_agent: str = DEFAULT_USER_AGENT,
    ) -> str:
        """Fetch `url` and return JSON with status, final URL, and body text."""
        try:
            response = await self._fetcher(url, timeout_seconds, user_agent)
        except httpx.HTTPError as exc:
            return json.dumps(
                {"error": "http_error", "url": url, "detail": str(exc)},
                indent=2,
            )
        text = html_to_text(response.text)[:MAX_BODY_CHARS]
        return json.dumps(
            {
                "url": url,
                "final_url": str(response.url),
                "status_code": response.status_code,
                "content_type": response.headers.get("content-type"),
                "text": text,
            },
            indent=2,
        )

    return [tool_from_function(async_fn=browse_url)]

html_to_text

html_to_text(html: str) -> str

Strip tags and return collapsed visible text.

Source code in mirai_shared_skills/browser/skill.py
def html_to_text(html: str) -> str:
    """Strip tags and return collapsed visible text."""
    parser = _TextExtractor()
    parser.feed(html)
    parser.close()
    return parser.text()

PDF Extraction

mirai_shared_skills.pdf_extraction

PDF Extraction foundational module.

__all__ module-attribute

__all__ = ['PdfExtractionSkill']

PdfExtractionSkill

Bases: BaseSkill

Standard skill that parses PDFs into plain text and metadata.

description property

description: str

instructions property

instructions: str

name property

name: str

get_tools

get_tools() -> list[Tool]
Source code in mirai_shared_skills/pdf_extraction/skill.py
def get_tools(self) -> list[Tool]:
    async def extract_pdf_metadata(path: str) -> str:
        """Return the document's page count, metadata fields, and encryption flag."""
        target = Path(path).expanduser()
        if not target.is_file():
            return json.dumps({"error": "not_a_file", "path": str(target)}, indent=2)
        payload = await asyncio.to_thread(_read_metadata, target)
        return json.dumps({"path": str(target), **payload}, indent=2)

    async def extract_pdf_text(path: str, max_pages: int = MAX_PAGES_PER_CALL) -> str:
        """Extract per-page text from `path`, truncated to `max_pages`."""
        target = Path(path).expanduser()
        if not target.is_file():
            return json.dumps({"error": "not_a_file", "path": str(target)}, indent=2)
        page_budget = max(1, min(max_pages, MAX_PAGES_PER_CALL))
        total_pages, pages = await asyncio.to_thread(_extract_pages, target, page_budget)
        joined = "\n\n".join(pages)
        truncated_chars = len(joined) > MAX_TEXT_CHARS
        return json.dumps(
            {
                "path": str(target),
                "total_pages": total_pages,
                "extracted_pages": len(pages),
                "truncated_chars": truncated_chars,
                "pages": [page[:MAX_TEXT_CHARS] for page in pages],
            },
            indent=2,
        )

    return [
        tool_from_function(async_fn=extract_pdf_metadata),
        tool_from_function(async_fn=extract_pdf_text),
    ]

skill

PDF Extraction — parse unstructured enterprise PDFs into LLM-ready text.

The skill wraps pypdf so the catalog stays dependency-light. Extraction is performed in a worker thread to keep the asyncio event loop responsive when an agent processes large documents.

MAX_PAGES_PER_CALL module-attribute

MAX_PAGES_PER_CALL: Final[int] = 100

MAX_TEXT_CHARS module-attribute

MAX_TEXT_CHARS: Final[int] = 64000

PdfExtractionSkill

Bases: BaseSkill

Standard skill that parses PDFs into plain text and metadata.

description property
description: str
instructions property
instructions: str
name property
name: str
get_tools
get_tools() -> list[Tool]
Source code in mirai_shared_skills/pdf_extraction/skill.py
def get_tools(self) -> list[Tool]:
    async def extract_pdf_metadata(path: str) -> str:
        """Return the document's page count, metadata fields, and encryption flag."""
        target = Path(path).expanduser()
        if not target.is_file():
            return json.dumps({"error": "not_a_file", "path": str(target)}, indent=2)
        payload = await asyncio.to_thread(_read_metadata, target)
        return json.dumps({"path": str(target), **payload}, indent=2)

    async def extract_pdf_text(path: str, max_pages: int = MAX_PAGES_PER_CALL) -> str:
        """Extract per-page text from `path`, truncated to `max_pages`."""
        target = Path(path).expanduser()
        if not target.is_file():
            return json.dumps({"error": "not_a_file", "path": str(target)}, indent=2)
        page_budget = max(1, min(max_pages, MAX_PAGES_PER_CALL))
        total_pages, pages = await asyncio.to_thread(_extract_pages, target, page_budget)
        joined = "\n\n".join(pages)
        truncated_chars = len(joined) > MAX_TEXT_CHARS
        return json.dumps(
            {
                "path": str(target),
                "total_pages": total_pages,
                "extracted_pages": len(pages),
                "truncated_chars": truncated_chars,
                "pages": [page[:MAX_TEXT_CHARS] for page in pages],
            },
            indent=2,
        )

    return [
        tool_from_function(async_fn=extract_pdf_metadata),
        tool_from_function(async_fn=extract_pdf_text),
    ]