.NET: Python/.Net: Agent Harness blog post accompanying samples part 3 (#6741)

* Python/.Net: Agent Harness blog post accompanying samples part 3

* Delete inadvertently added files

* Address PR feedback.

* Rename files that are causing dotnet format failures

* Address PR comment

* Fix blog links
This commit is contained in:
westey
2026-07-09 14:52:53 +01:00
committed by GitHub
parent f5c078b70f
commit fbaa346eec
38 changed files with 1710 additions and 0 deletions
@@ -7,6 +7,7 @@ directory.
- **Part 1 — `claw_step01_meet_your_claw.py`** — the minimal harness.
- **Part 2 — `claw_step02_working_with_data.py`** — file access, approvals, and durable memory.
- **Part 3 — `claw_step03_scaling_capabilities.py`** — skills, shell, CodeAct, and background agents.
## Prerequisites
@@ -116,3 +117,61 @@ watchlist) lives on disk keyed by session id, so `/session-export` before you qu
`/session-import` after relaunching to re-link the relaunched session to its files, then ask
*"What's on my watchlist?"* or *"What do you know about me?"*.
## Part 3 — Scaling its capabilities
Makes the assistant *more capable* along four axes.
### What this sample demonstrates
- **Skills** — finance know-how (`valuation`, `risk-scoring`) is packaged as discoverable
`SKILL.md` files under `skills/`, which the agent loads on demand. The sample builds a
`FileSkillsSource(..., script_runner=subprocess_script_runner)` so the skills' Python
scripts can run. Optionally folds in centrally-managed **Foundry skills** served from a
Foundry Toolbox MCP endpoint via `MCPSkillsSource` (opt-in; see below).
- **Shell** — a `LocalShellTool` confined to the trade-confirmation vault
(`working/confirmations/`) lets the agent tidy the accumulated confirmation files (reorganize into
`year/month`, rename to `YYYY-MM-DD_TICKER_BUY|SELL.txt`). Guarded by a `ShellPolicy` deny-list
**and** a confined working directory; left at the default
`approval_mode="always_require"` so each command is surfaced for approval.
- **CodeAct** — a `MontyCodeActProvider` gives the agent a sandboxed, cross-platform Python
interpreter to crunch portfolio numbers by writing and running code.
- **Background agents** — a lean, web-search-only `TickerResearchAgent` is registered via
`create_harness_agent(background_agents=[...])`, so the main agent can fan out per-ticker research
concurrently and aggregate the findings.
### Additional environment variables (optional)
```bash
# Enable centrally-managed Foundry skills (Foundry Toolbox MCP endpoint URL):
export FOUNDRY_TOOLBOX_MCP_SERVER_URL="https://<your-project>.services.ai.azure.com/.../toolboxes/<toolbox>/mcp?api-version=v1"
```
When this is not set, the sample runs with the local file skills only, and prints a note.
### Running
```bash
uv run python/samples/02-agents/harness/build_your_own_claw/claw_step03_scaling_capabilities.py
```
### What to expect
Try these in order (the sample starts in **execute** mode — quick lookups don't need a plan):
1. `Value MSFT for me.` — the agent loads the `valuation` skill and follows its instructions
(reading references and running its script).
2. `Score the risk of my portfolio.` — the agent reads `portfolio.csv` and loads the `risk-scoring`
skill.
3. `/mode plan`, then `Tidy up my trade confirmations.` — switching to plan mode first makes the
agent inspect `working/confirmations/` and propose a reorganization plan before touching anything;
once you approve it switches to execute and uses the shell to reorganize and rename the files,
**prompting you to approve** each command.
4. `Work out the total value of my portfolio.` — the agent writes and runs Python via CodeAct.
5. `Research MSFT, NVDA and SPY and summarize the latest news.` — the agent fans the tickers out to
the background research agent and aggregates the results.
6. `What's the capital of France?` — with a `financial-agent-rules` skill published to your Foundry
toolbox and Foundry skills enabled (`FOUNDRY_TOOLBOX_MCP_SERVER_URL`), the agent loads it,
recognizes the question is off-topic, and politely declines, steering you back to finance.
See the [Part 3 blog post](https://devblogs.microsoft.com/agent-framework/agent-harness-scaling-the-claw-or-harness-capabilities/)
for more on the `financial-agent-rules` skill — including the SKILL.md to publish to your Foundry toolbox.
@@ -0,0 +1,339 @@
# /// script
# requires-python = ">=3.10"
# dependencies = [
# "agent-framework",
# "agent-framework-tools",
# "agent-framework-monty",
# "mcp",
# "httpx",
# "textual>=6.2.1",
# "rich>=13.7.1",
# "azure-identity",
# "python-dotenv",
# ]
# ///
# Run with any PEP 723 compatible runner, e.g.:
# uv run python/samples/02-agents/harness/build_your_own_claw/claw_step03_scaling_capabilities.py
# Copyright (c) Microsoft. All rights reserved.
"""Scaling its capabilities (Post 3) — Python.
The third runnable sample from the "Build your own claw and agent harness with Microsoft Agent
Framework" blog series. See: https://devblogs.microsoft.com/agent-framework/agent-harness-scaling-the-claw-or-harness-capabilities/.
It builds on Post 2's personal finance assistant and makes it *more capable* in four ways:
1. Skills — package finance know-how (valuation, risk-scoring) as discoverable SKILL.md
files the agent loads on demand. Optionally fold in centrally-managed Foundry
skills served from a Foundry Toolbox MCP endpoint (opt-in via
FOUNDRY_TOOLBOX_MCP_SERVER_URL).
2. Shell — a sandboxed shell, confined to the trade-confirmation vault, that the agent uses
to reorganize the accumulated confirmation files (year/month, rename, archive).
Guarded by an allow/deny-list policy and a confined working directory.
3. CodeAct — the agent writes and runs Python to crunch portfolio numbers, using the
cross-platform Monty interpreter.
4. Background agents — fan out a per-ticker research sub-agent so several tickers are researched
concurrently, then aggregated.
This sample reuses the shared harness ``console`` package in the parent ``harness/`` directory.
Environment variables:
FOUNDRY_PROJECT_ENDPOINT — Microsoft Foundry project endpoint URL
FOUNDRY_MODEL — Model deployment name (defaults to gpt-5.4)
FOUNDRY_TOOLBOX_MCP_SERVER_URL — (optional) Foundry Toolbox MCP endpoint URL; enables Foundry skills
Authentication:
Run ``az login`` before running this sample.
"""
import asyncio
import os
import sys
import uuid
from collections.abc import Callable, Generator
from contextlib import AsyncExitStack
from datetime import datetime, timezone
from pathlib import Path
from typing import Annotated, Any, Literal
import httpx
from agent_framework import (
AggregatingSkillsSource,
Agent,
AgentModeProvider,
DeduplicatingSkillsSource,
FileAccessProvider,
FileSkillsSource,
FileSystemAgentFileStore,
MCPSkillsSource,
SkillsProvider,
SkillsSource,
create_harness_agent,
tool,
)
from agent_framework.foundry import FoundryChatClient
from agent_framework_monty import MontyCodeActProvider
from agent_framework_tools.shell import LocalShellTool, ShellPolicy
from azure.identity import AzureCliCredential, get_bearer_token_provider
from dotenv import load_dotenv
from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamable_http_client
from pydantic import Field
# Reuse the shared harness console that lives in the parent ``harness/`` directory, and the local
# subprocess script runner used to execute file-based skill scripts.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from console import build_observers_with_planning, run_agent_async # noqa: E402
from subprocess_script_runner import subprocess_script_runner # noqa: E402
_SAMPLE_DIR = Path(__file__).resolve().parent
_WORKING_DIR = _SAMPLE_DIR / "working"
_VAULT_DIR = _WORKING_DIR / "confirmations"
_SKILLS_DIR = _SAMPLE_DIR / "skills"
FINANCE_INSTRUCTIONS = """\
## Personal Finance Assistant Instructions
You are a personal finance and investing assistant. You help the user understand their portfolio
and watchlist, value individual stocks, gauge portfolio risk, research the market, and keep their
records tidy.
### Working style
- The user's holdings live in a file called portfolio.csv. Read it with the file_access tools
before answering questions about their portfolio, and never modify it unless asked.
- You have skills for valuation and risk-scoring. When a question matches a skill, load it and
follow its instructions (read its references, run its scripts) rather than guessing.
- When asked to research several tickers, delegate each one to the background research agent so
they run concurrently, then summarize the findings together.
- The user's trade confirmations accumulate in the working/confirmations folder. When asked to tidy
or reorganize them, use the run_shell tool: inspect the folder first, then move files into a
year/month layout and rename them to YYYY-MM-DD_TICKER_BUY|SELL.txt. Explain your plan before
running commands that change anything.
- To buy or sell, use the place_trade tool. This takes a real action, so the user will be asked to
approve it before it runs — explain what you are about to do first.
### Important
You provide information and analysis only — you are not a licensed financial advisor and you must
not present your output as personalized investment advice. Remind the user to do their own
research before making decisions.
"""
# A tiny in-memory book of (price, trailing EPS) so the sample runs without any external dependency.
# These are illustrative mock values, not real market data.
_PRICE_BOOK: dict[str, tuple[float, float]] = {
"MSFT": (462.97, 11.80),
"AAPL": (229.35, 6.13),
"GOOGL": (178.12, 7.54),
"AMZN": (201.45, 4.18),
"NVDA": (134.81, 2.95),
"SPY": (612.40, 23.10),
}
# <get_stock_price>
def get_stock_price(
symbol: Annotated[str, "The stock ticker symbol, e.g. MSFT or AAPL."],
) -> dict[str, object]:
"""Get the latest (delayed, illustrative) stock price and trailing EPS for a ticker symbol."""
ticker = symbol.upper()
data = _PRICE_BOOK.get(ticker)
if data is None:
# Deterministic pseudo-values for unknown symbols so the sample stays self-contained.
# The built-in hash() is randomized per process (PYTHONHASHSEED), so derive a stable seed.
seed = 0
for ch in ticker:
seed = (seed * 31 + ord(ch)) % 1_000_000
price = 50.0 + (seed % 45000) / 100.0
data = (price, round(price / 20.0, 2))
return {
"symbol": ticker,
"price": round(data[0], 2),
"trailing_eps": round(data[1], 2),
"currency": "USD",
"as_of": datetime.now(timezone.utc).isoformat(),
}
# </get_stock_price>
# <place_trade>
@tool(approval_mode="always_require")
def place_trade(
symbol: Annotated[str, "The stock ticker symbol to trade, e.g. MSFT."],
action: Annotated[Literal["buy", "sell"], "Either 'buy' or 'sell'."],
quantity: Annotated[int, Field(gt=0, description="The number of shares to trade.")],
) -> str:
"""Place a (simulated) buy or sell order. Marked approval-required, so the harness asks the
user to approve before this ever runs. No real order is placed.
``action`` and ``quantity`` are validated by the framework (pydantic) from their type hints:
the model can only pass 'buy'/'sell' and a quantity greater than zero.
"""
verb = "Sold" if action == "sell" else "Bought"
confirmation = f"TRADE-{uuid.uuid4().hex[:8].upper()}"
return f"{verb} {quantity} share(s) of {symbol.upper()}. Confirmation: {confirmation}."
# </place_trade>
# <skills>
async def _build_skills_provider(stack: AsyncExitStack) -> SkillsProvider:
"""Build a skills provider over the local skills/ folder, plus optional Foundry-managed skills.
File-based skills (valuation, risk-scoring) always load. When FOUNDRY_TOOLBOX_MCP_SERVER_URL is
set we also connect to a Foundry Toolbox MCP endpoint and surface its skills, so they can be
managed and updated centrally without changing this agent.
"""
# subprocess_script_runner lets the file-based skills run their Python scripts.
sources: list[SkillsSource] = [FileSkillsSource(str(_SKILLS_DIR), script_runner=subprocess_script_runner)]
toolbox_url = os.environ.get("FOUNDRY_TOOLBOX_MCP_SERVER_URL")
if toolbox_url:
session = await _connect_foundry_toolbox(stack, toolbox_url)
sources.append(MCPSkillsSource(client=session))
print("Foundry skills enabled (Toolbox MCP).")
else:
print("Foundry skills disabled. Set FOUNDRY_TOOLBOX_MCP_SERVER_URL to enable them.")
source: SkillsSource = sources[0] if len(sources) == 1 else AggregatingSkillsSource(sources)
return SkillsProvider(DeduplicatingSkillsSource(source))
class _ToolboxAuth(httpx.Auth):
"""Attach a fresh Foundry bearer token to every request."""
def __init__(self, token_provider: Callable[[], str]):
self._get_token = token_provider
def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, None]:
request.headers["Authorization"] = f"Bearer {self._get_token()}"
yield request
async def _connect_foundry_toolbox(stack: AsyncExitStack, url: str) -> ClientSession:
"""Open an MCP session against a Foundry Toolbox endpoint, tied to ``stack``'s lifetime."""
token_provider = get_bearer_token_provider(AzureCliCredential(), "https://ai.azure.com/.default")
http_client = await stack.enter_async_context(
httpx.AsyncClient(
auth=_ToolboxAuth(token_provider),
headers={"Foundry-Features": "Toolboxes=V1Preview"},
timeout=httpx.Timeout(30.0, read=300.0),
follow_redirects=True,
)
)
read, write, _ = await stack.enter_async_context(streamable_http_client(url=url, http_client=http_client))
session = await stack.enter_async_context(ClientSession(read, write))
await session.initialize()
return session
# </skills>
# <background>
def _build_research_agent(client: FoundryChatClient) -> Any:
"""Build the lean, web-search-only chat agent used for per-ticker research."""
# This sub-agent doesn't need any harness machinery - it's a plain chat agent with a single
# tool: the same hosted web search the harness would have added. The parent still exposes the
# background_agents_* tools because it receives this agent via background_agents.
return Agent(
client=client,
name="TickerResearchAgent",
description="Searches the web for recent news and commentary about a single stock ticker.",
tools=[client.get_web_search_tool()],
instructions=(
"You research a single stock ticker. Use the web search tool to find the most recent, "
"relevant news and commentary, then return a short, factual summary (3-4 bullet points) "
"with no preamble."
),
)
# </background>
# <shell>
def _build_shell() -> LocalShellTool:
"""A sandboxed shell, confined to the trade-confirmation vault.
``confine_workdir`` re-anchors every command to the vault, and the deny-list pre-filters
obviously destructive command shapes. (Patterns are a UX guardrail, not a security boundary —
for hard isolation use DockerShellTool.) Left at the default ``approval_mode="always_require"``
so each command is surfaced for approval.
"""
return LocalShellTool(
mode="persistent",
workdir=str(_VAULT_DIR),
confine_workdir=True,
policy=ShellPolicy(
denylist=[
r"\brm\s+-rf\b",
r"\bsudo\b",
r":\(\)\s*\{", # fork-bomb shape
r"\bmkfs\b",
r">\s*/dev/sd",
],
),
timeout=15,
)
# </shell>
async def main() -> None:
load_dotenv()
_WORKING_DIR.mkdir(exist_ok=True)
# <create_client>
# Construct a chat client (see Post 1). FoundryChatClient reads FOUNDRY_PROJECT_ENDPOINT and
# FOUNDRY_MODEL from the environment; AzureCliCredential handles auth (run `az login`).
client = FoundryChatClient(credential=AzureCliCredential())
# </create_client>
async with AsyncExitStack() as stack:
skills_provider = await _build_skills_provider(stack)
research_agent = _build_research_agent(client)
shell = _build_shell()
# <codeact>
# CodeAct: a sandboxed Python interpreter the model can write and run code in to crunch
# numbers. Monty is a pure, cross-platform interpreter, so it needs no extra setup.
context_providers: list[Any] = [MontyCodeActProvider(approval_mode="never_require")]
print("CodeAct enabled (Monty).")
# </codeact>
# <create_agent>
# Turn the chat client into a harness agent. On top of Post 2's file access and approvals we
# add the four "scaling" capabilities: skills (our own provider), background agents, a
# confined shell, and optional CodeAct. Read-only file tools are auto-approved so reading the
# portfolio is frictionless while writes, trades, and shell commands still prompt.
agent = create_harness_agent(
client=client,
agent_instructions=FINANCE_INSTRUCTIONS,
tools=[get_stock_price, place_trade],
file_access_store=FileSystemAgentFileStore(str(_WORKING_DIR)),
skills_provider=skills_provider,
background_agents=[research_agent],
shell_executor=shell,
auto_approval_rules=[FileAccessProvider.read_only_tools_auto_approval_rule],
context_providers=context_providers,
mode_provider=AgentModeProvider(default_mode="execute"),
)
# </create_agent>
# <run>
session = agent.create_session()
# Run the interactive console session. The default planning observers already include a tool
# approval observer, so the place_trade and run_shell approval prompts are surfaced
# automatically.
await run_agent_async(
agent,
session=session,
observers=build_observers_with_planning(agent),
initial_mode="execute",
title="💹 Finance Assistant",
placeholder="Value a stock, score your portfolio risk, research tickers, or tidy your confirmations...",
)
# </run>
if __name__ == "__main__":
asyncio.run(main())
@@ -0,0 +1,18 @@
---
name: risk-scoring
description: Score how concentrated and risky a portfolio is on a 0-100 scale from its position weights. Use when the user asks how risky their portfolio is, whether it is too concentrated, or for a diversification check.
---
## Usage
When the user asks about portfolio risk or concentration:
1. Read `references/risk-bands.md` to understand the score bands and what drives them.
2. Compute each holding's market value (shares × price) — use the `get_stock_price` tool for current
prices if you do not already have them.
3. Run `scripts/risk_score.py` with one `--position VALUE` argument per holding,
e.g. `--position 18518 --position 17201 --position 16177`.
4. Report the 0-100 score, the band it falls in, and the largest single-position weight, then suggest
(in general terms) whether the portfolio looks well diversified or concentrated.
Remind the user this is a crude concentration measure, not a complete risk model, and not advice.
@@ -0,0 +1,27 @@
# Risk-scoring guide (illustrative)
This skill scores **concentration risk** — how much a portfolio depends on its largest positions —
on a 0-100 scale, where higher means riskier.
## How the score is built
1. Convert each position to a weight: `weight = position_value / total_value`.
2. Compute the Herfindahl-Hirschman Index (HHI): `HHI = sum(weight^2)`.
- A perfectly even portfolio of *n* holdings has `HHI = 1/n` (low).
- A single-stock portfolio has `HHI = 1` (maximum concentration).
3. Scale to 0-100: `score = round(HHI * 100)`.
## Score bands
| Score | Band | Interpretation |
|---------|--------------------|-------------------------------------------------|
| 0-20 | Well diversified | No single holding dominates. |
| 21-40 | Moderately diversified | Some tilt, but broadly spread. |
| 41-60 | Concentrated | A few positions carry most of the risk. |
| 61-100 | Highly concentrated| Heavily dependent on one or two positions. |
Also watch the **largest single-position weight**: above ~25% is usually worth flagging regardless
of the overall score.
This measures concentration only — it ignores volatility, correlation, sector exposure, and leverage,
so it is a starting point, not a verdict.
@@ -0,0 +1,58 @@
# Portfolio risk-scoring script
# Scores concentration risk on a 0-100 scale using the Herfindahl-Hirschman Index (HHI).
#
# weight_i = position_i / total
# HHI = sum(weight_i ^ 2)
# score = round(HHI * 100) # higher = more concentrated = riskier
#
# Usage:
# python scripts/risk_score.py --position 18518 --position 17201 --position 16177
import argparse
import json
def main() -> None:
parser = argparse.ArgumentParser(description="Score portfolio concentration risk (0-100).")
parser.add_argument(
"--position",
type=float,
action="append",
required=True,
help="Market value of one holding. Pass once per position.",
)
args = parser.parse_args()
positions = args.position
if any(p <= 0 for p in positions):
print(json.dumps({"error": "Each position value must be a positive market value."}))
return
total = sum(positions)
if total <= 0:
print(json.dumps({"error": "Total portfolio value must be positive."}))
return
weights = [p / total for p in positions]
hhi = sum(w * w for w in weights)
score = round(hhi * 100)
if score <= 20:
band = "Well diversified"
elif score <= 40:
band = "Moderately diversified"
elif score <= 60:
band = "Concentrated"
else:
band = "Highly concentrated"
print(json.dumps({
"positions": len(positions),
"score": score,
"band": band,
"largest_weight_pct": round(max(weights) * 100, 1),
}))
if __name__ == "__main__":
main()
@@ -0,0 +1,17 @@
---
name: valuation
description: Estimate whether a stock looks cheap or expensive using a price-to-earnings (P/E) based fair-value method. Use when the user asks if a stock is over- or under-valued, or for a fair-value / target price.
---
## Usage
When the user asks whether a stock is fairly valued, over-valued, or under-valued:
1. Read `references/valuation-guide.md` to pick a sensible target P/E for the company's sector.
2. Run `scripts/valuation_metrics.py` with the current price, trailing EPS, and the target P/E,
e.g. `--price 462.97 --eps 11.80 --target-pe 32`.
3. Report the computed P/E, the fair-value estimate, and the percentage upside/downside, then state
plainly whether the stock looks cheap or expensive on this measure.
Always remind the user that a single P/E heuristic is not investment advice and ignores growth,
debt, and many other factors.
@@ -0,0 +1,28 @@
# Valuation guide (illustrative)
A quick price-to-earnings (P/E) sanity check:
- **P/E = price ÷ trailing earnings per share (EPS)**
- **Fair value = trailing EPS × target P/E**
- **Upside/downside = (fair value price) ÷ price**
## Typical target P/E by sector
These are rough, illustrative anchors only — not live market multiples.
| Sector | Conservative target P/E | Growth target P/E |
|-----------------------|-------------------------|-------------------|
| Mega-cap technology | 28 | 35 |
| Semiconductors | 25 | 40 |
| Consumer staples | 18 | 22 |
| Financials / banks | 11 | 14 |
| Broad market (index) | 19 | 21 |
## How to read the result
- Fair value **well above** the current price ⇒ the stock looks **cheap** on this measure.
- Fair value **well below** the current price ⇒ the stock looks **expensive** on this measure.
- Within ~5% ⇒ roughly **fairly valued**.
This is one crude lens. It ignores growth rates, balance-sheet strength, and cash flow, so never
present it as a recommendation.
@@ -0,0 +1,57 @@
# Valuation metrics script
# Computes a simple price-to-earnings (P/E) based fair-value estimate.
#
# fair_value = eps * target_pe
# pe = price / eps
# upside = (fair_value - price) / price
#
# Usage:
# python scripts/valuation_metrics.py --price 462.97 --eps 11.80 --target-pe 32
import argparse
import json
def main() -> None:
parser = argparse.ArgumentParser(description="Compute a P/E based fair-value estimate.")
parser.add_argument("--price", type=float, required=True, help="Current share price.")
parser.add_argument("--eps", type=float, required=True, help="Trailing earnings per share.")
parser.add_argument("--target-pe", type=float, required=True, help="Target P/E from the guide.")
args = parser.parse_args()
if args.eps <= 0:
print(json.dumps({"error": "EPS must be positive to compute a P/E ratio."}))
return
if args.price <= 0:
print(json.dumps({"error": "Price must be positive to compute valuation metrics."}))
return
if args.target_pe <= 0:
print(json.dumps({"error": "Target P/E must be positive."}))
return
pe = args.price / args.eps
fair_value = args.eps * args.target_pe
upside = (fair_value - args.price) / args.price
if upside > 0.05:
verdict = "looks cheap"
elif upside < -0.05:
verdict = "looks expensive"
else:
verdict = "roughly fairly valued"
print(json.dumps({
"price": round(args.price, 2),
"eps": round(args.eps, 2),
"target_pe": round(args.target_pe, 2),
"pe": round(pe, 2),
"fair_value": round(fair_value, 2),
"upside_pct": round(upside * 100, 1),
"verdict": verdict,
}))
if __name__ == "__main__":
main()
@@ -0,0 +1,77 @@
# Copyright (c) Microsoft. All rights reserved.
"""Sample subprocess-based skill script runner.
Executes file-based skill scripts as local Python subprocesses.
This is provided for demonstration purposes only.
"""
from __future__ import annotations
import subprocess
import sys
# Uncomment this filter to suppress the experimental Skills warning before
# using the sample's Skills APIs.
# import warnings
# warnings.filterwarnings("ignore", message=r"\[SKILLS\].*", category=FutureWarning)
from pathlib import Path
from typing import Any
from agent_framework import FileSkill, FileSkillScript
def subprocess_script_runner(
skill: FileSkill, script: FileSkillScript, args: dict[str, Any] | list[str] | None = None
) -> str:
"""Run a skill script as a local Python subprocess.
Uses ``FileSkillScript.full_path`` as the script path, converts the
``args`` to CLI arguments, and returns captured output.
Args:
skill: The file-based skill that owns the script.
script: The file-based script to run.
args: Optional arguments. A ``list[str]`` is forwarded as
positional CLI arguments. Passing a ``dict`` or any other
type raises :class:`TypeError` — file-based scripts expect
positional arguments as a JSON array of strings.
Returns:
The combined stdout/stderr output, or an error message.
Raises:
TypeError: If ``args`` is not a ``list[str]`` or ``None``, or if
any list element is not a string.
"""
script_path = Path(script.full_path)
if not script_path.is_file():
return f"Error: Script file not found: {script_path}"
cmd = [sys.executable, str(script_path)]
if isinstance(args, list):
for item in args:
if not isinstance(item, str):
raise TypeError(
f"File-based skill scripts only accept string CLI arguments "
f"but received a {type(item).__name__}. "
f"All array elements must be strings."
)
cmd.extend(args)
elif args is not None:
raise TypeError(
f"Expected a list of CLI arguments but received {type(args).__name__}. "
f"File-based skill scripts expect positional arguments as a list of strings."
)
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30,
cwd=str(script_path.parent),
)
output = result.stdout
if result.stderr:
output += f"\nStderr:\n{result.stderr}"
if result.returncode != 0:
output += f"\nScript exited with code {result.returncode}"
return output.strip() or "(no output)"
except subprocess.TimeoutExpired:
return f"Error: Script '{script.name}' timed out after 30 seconds."
except OSError as e:
return f"Error: Failed to execute script '{script.name}': {e}"
@@ -0,0 +1,6 @@
TRADE CONFIRMATION
Confirmation: TRADE-55AA44BB
Date: 2025-06-21
Symbol: NVDA
Action: SELL
Quantity: 20
@@ -0,0 +1,6 @@
TRADE CONFIRMATION
Confirmation: TRADE-77CC88DD
Date: 2024-05-08
Symbol: SPY
Action: SELL
Quantity: 15
@@ -0,0 +1,6 @@
TRADE CONFIRMATION
Confirmation: TRADE-9F8E7D6C
Date: 2024-11-03
Symbol: AAPL
Action: BUY
Quantity: 75
@@ -0,0 +1,6 @@
TRADE CONFIRMATION
Confirmation: TRADE-1234ABCD
Date: 2025-09-12
Symbol: AMZN
Action: BUY
Quantity: 30
@@ -0,0 +1,6 @@
TRADE CONFIRMATION
Confirmation: TRADE-EE11FF22
Date: 2025-01-30
Symbol: GOOGL
Action: BUY
Quantity: 25
@@ -0,0 +1,6 @@
TRADE CONFIRMATION
Confirmation: TRADE-A1B2C3D4
Date: 2024-02-14
Symbol: MSFT
Action: BUY
Quantity: 40