repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
Mirrored from https://github.com/yingqi-z20/Agent-libOS
stars
latest
clone command
git clone gitlawb://did:key:z6MkqRzA...RfoM/yingqi-z20-Agen...git clone gitlawb://did:key:z6MkqRzA.../yingqi-z20-Agen...d98dd2c9IPC1d ago| #1 | from __future__ import annotations |
| #2 | |
| #3 | import re |
| #4 | from typing import Any |
| #5 | |
| #6 | from pydantic import BaseModel, ConfigDict, Field |
| #7 | |
| #8 | from agent_libos.config import DEFAULT_CONFIG |
| #9 | from agent_libos.tools.base import SyncAgentTool, ToolContext, ToolPolicy |
| #10 | |
| #11 | _TOOL_DEFAULTS = DEFAULT_CONFIG.tools |
| #12 | |
| #13 | |
| #14 | class EchoArgs(BaseModel): |
| #15 | model_config = ConfigDict(extra="allow") |
| #16 | |
| #17 | |
| #18 | class EchoTool(SyncAgentTool[EchoArgs]): |
| #19 | name = "echo" |
| #20 | description = "Return the provided arguments unchanged. Useful for tool plumbing tests." |
| #21 | args_schema = EchoArgs |
| #22 | policy = ToolPolicy(side_effects=False, idempotent=True, timeout_s=_TOOL_DEFAULTS.interactive_timeout_s) |
| #23 | tags = ["debug", "deterministic"] |
| #24 | |
| #25 | def run(self, args: EchoArgs, ctx: ToolContext) -> dict[str, Any]: |
| #26 | return args.model_dump() |
| #27 | |
| #28 | |
| #29 | class ParsePytestLogArgs(BaseModel): |
| #30 | log: str = Field(description="Raw pytest output.") |
| #31 | |
| #32 | |
| #33 | class ParsePytestLogOutput(BaseModel): |
| #34 | failed: list[str] |
| #35 | errors: list[str] |
| #36 | assertions: list[str] |
| #37 | failure_count: int |
| #38 | |
| #39 | |
| #40 | class ParsePytestLogTool(SyncAgentTool[ParsePytestLogArgs]): |
| #41 | name = "parse_pytest_log" |
| #42 | description = "Parse pytest output into a small structured failure summary." |
| #43 | args_schema = ParsePytestLogArgs |
| #44 | output_schema = ParsePytestLogOutput |
| #45 | policy = ToolPolicy(side_effects=False, idempotent=True, timeout_s=_TOOL_DEFAULTS.interactive_timeout_s) |
| #46 | tags = ["coding", "pytest", "parser"] |
| #47 | |
| #48 | def run(self, args: ParsePytestLogArgs, ctx: ToolContext) -> ParsePytestLogOutput: |
| #49 | failed: list[str] = [] |
| #50 | errors: list[str] = [] |
| #51 | assertions: list[str] = [] |
| #52 | for line in args.log.splitlines(): |
| #53 | stripped = line.strip() |
| #54 | if stripped.startswith("FAILED "): |
| #55 | failed.append(stripped) |
| #56 | elif re.match(r"^E\s+", stripped): |
| #57 | errors.append(stripped[2:]) |
| #58 | elif "AssertionError" in stripped: |
| #59 | assertions.append(stripped) |
| #60 | return ParsePytestLogOutput( |
| #61 | failed=failed, |
| #62 | errors=errors, |
| #63 | assertions=assertions, |
| #64 | failure_count=len(failed) or len(assertions) or len(errors), |
| #65 | ) |
| #66 |