#4906 Python Skill to Agent
#4906 Python Skill to Agent
#4906 Provide python code to turn a selected skill.md file and pydantic_classes.py file into an executable AI agent
An executable AI agent that reads instructions directly from a structured skill.md file and dynamically binds tools from a pydantic_classes.py schema can be built cleanly using Python, Pydantic, and an LLM framework (such as the standard openai SDK with structured function calling).
Project Layout
agent_project/
├── skill.md
├── pydantic_classes.py
└── agent.py
1. pydantic_classes.py
Define the input schemas for any tools your agent can call.
from pydantic import BaseModel, Field
class WebSearch(BaseModel):
"""Search the web for up-to-date facts, documentation, or articles."""
query: str = Field(..., description="The search query string")
max_results: int = Field(
default=3, description="Number of results to return"
)
class ExecutePython(BaseModel):
"""Execute arbitrary Python code and return the standard output."""
code: str = Field(..., description="Valid Python 3 source code to execute")
2. skill.md
The skill definition document outlining agent persona, directives, and operational workflow.
# Skill: Data Assistant & Researcher
## Role
You are an analytical assistant that combines live research with Python computation to solve problems accurately.
## Instructions
1. Always verify computations by executing Python code.
2. Formulate concise search queries when external context is required.
3. Present final answers directly with clear formatting.
3. agent.py
This script parses skill.md, introspects pydantic_classes.py to convert every subclass of BaseModel into an OpenAI-compatible tool definition, and runs an execution loop handling multi-step tool calls.
import importlib.util
import inspect
import json
import sys
from pathlib import Path
from typing import Any, Callable, Dict, List, Type
from openai import OpenAI
from pydantic import BaseModel
def load_skill_content(skill_path: str | Path) -> str:
"""Reads instructions directly from a skill.md file."""
path = Path(skill_path)
if not path.is_file():
raise FileNotFoundError(f"Skill file not found: {path}")
return path.read_text(encoding="utf-8")
def load_pydantic_models(
module_path: str | Path,
) -> Dict[str, Type[BaseModel]]:
"""Dynamically loads all Pydantic BaseModel subclasses defined in a module file."""
path = Path(module_path).resolve()
if not path.is_file():
raise FileNotFoundError(f"Pydantic module not found: {path}")
spec = importlib.util.spec_from_file_location("dynamic_tools", path)
if spec is None or spec.loader is None:
raise ImportError(f"Could not load module spec for: {path}")
module = importlib.util.module_from_spec(spec)
sys.modules["dynamic_tools"] = module
spec.loader.exec_module(module)
models: Dict[str, Type[BaseModel]] = {}
for name, obj in inspect.getmembers(module, inspect.isclass):
if issubclass(obj, BaseModel) and obj is not BaseModel:
models[name] = obj
return models
def pydantic_to_openai_tool(model: Type[BaseModel]) -> dict:
"""Converts a Pydantic model into OpenAI function/tool schema."""
schema = model.model_json_schema()
return {
"type": "function",
"function": {
"name": model.__name__,
"description": model.__doc__ or schema.get("description", ""),
"parameters": {
"type": "object",
"properties": schema.get("properties", {}),
"required": schema.get("required", []),
},
},
}
# =====================================================================
# Mock or Real Implementations for Tool Execution
# =====================================================================
def mock_web_search(query: str, max_results: int = 3) -> str:
return json.dumps(
[{"title": f"Result for '{query}'", "snippet": "Sample data snippet"}]
)
def mock_execute_python(code: str) -> str:
import io
import sys
buffer = io.StringIO()
old_stdout = sys.stdout
try:
sys.stdout = buffer
exec(code, {})
output = buffer.getvalue()
return output if output else "[Code executed successfully with no output]"
except Exception as e:
return f"Execution Error: {e}"
finally:
sys.stdout = old_stdout
TOOL_DISPATCH: Dict[str, Callable[..., Any]] = {
"WebSearch": mock_web_search,
"ExecutePython": mock_execute_python,
}
# =====================================================================
# Executable Agent Loop
# =====================================================================
class ExecutableSkillAgent:
def __init__(
self,
skill_file: str = "skill.md",
pydantic_file: str = "pydantic_classes.py",
model: str = "gpt-4o",
):
self.client = OpenAI()
self.model = model
self.system_prompt = load_skill_content(skill_file)
self.models = load_pydantic_models(pydantic_file)
self.tools = [
pydantic_to_openai_tool(model_cls)
for model_cls in self.models.values()
]
def run(self, user_prompt: str) -> str:
messages: List[Dict[str, Any]] = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_prompt},
]
while True:
response = self.client.chat.completions.create(
model=self.model, messages=messages, tools=self.tools or None
)
msg = response.choices[0].message
messages.append(msg.model_dump(exclude_unset=True))
if not msg.tool_calls:
return msg.content or ""
for call in msg.tool_calls:
fn_name = call.function.name
arguments = json.loads(call.function.arguments)
# Validate payload using the dynamically loaded Pydantic class
model_cls = self.models.get(fn_name)
if model_cls:
validated_payload = model_cls(**arguments)
handler = TOOL_DISPATCH.get(fn_name)
if handler:
result = handler(**validated_payload.model_dump())
else:
result = f"Error: No implementation handler registered for {fn_name}"
else:
result = f"Error: Unknown tool schema {fn_name}"
messages.append(
{
"role": "tool",
"tool_call_id": call.id,
"content": str(result),
}
)
if __name__ == "__main__":
agent = ExecutableSkillAgent(
skill_file="skill.md",
pydantic_file="pydantic_classes.py",
)
query = "Calculate the sum of all prime numbers under 50 using Python."
print(f"User: {query}\n")
response = agent.run(query)
print(f"Agent:\n{response}")
Key Elements of the Setup
Dynamic Loading via
importlib: Avoids hardcoding class names; any newBaseModeladded topydantic_classes.pyautomatically registers as a callable agent tool.Schema Generation:
model.model_json_schema()converts Pydantic fields, types, constraints, and descriptions into strict JSON schemas for the API function definitions.Payload Validation: Incoming LLM arguments are parsed back into the Pydantic model (
model_cls(**arguments)) before execution, catching malformed outputs before they reach the tool handlers.
What is decorator-based tool dispatch, with binding?
Decorator-based tool dispatch with binding is a software design pattern where Python decorators mark functions as agent tools and automatically bind their execution logic, name, metadata, and validation schemas (often via Pydantic or type annotations) into an executable registry.
Instead of writing separate lists of schemas and manually writing if/else or match blocks to route function calls, the decorator handles registration, schema extraction, and argument execution in one place.
Core Mechanics
Decoration & Registration: A custom decorator wraps a standard Python function. During module load, it registers the function in a tool registry.
Schema Extraction: The decorator inspects type hints, docstrings, or associated Pydantic models to produce the JSON schema required by the model (e.g., OpenAI/Anthropic tool schemas).
Dispatch & Execution (Binding): When the model selects a tool by name, the registry looks up the function, parses and validates the arguments against the schema, binds any dependencies (like state or API keys), and executes the function.
End-to-End Implementation Example
Here is a minimal, fully functional implementation of a decorator-based tool registry with Pydantic validation and automatic dispatch:
import inspect
import json
from functools import wraps
from typing import Any, Callable, Dict, Type
from pydantic import BaseModel, Field, create_model
class ToolRegistry:
def __init__(self):
# Maps tool name -> metadata & executable handler
self._tools: Dict[str, Dict[str, Any]] = {}
def register(self, schema_model: Type[BaseModel] | None = None):
"""Decorator to register a function as an agent tool, binding it to a schema."""
def decorator(func: Callable[..., Any]):
tool_name = func.__name__
description = func.__doc__ or "No description provided."
# Use the explicit Pydantic model if provided,
# otherwise synthesize one from function type hints
if schema_model is not None:
model = schema_model
else:
# Dynamically construct a Pydantic model from function signature
sig = inspect.signature(func)
fields = {}
for param_name, param in sig.parameters.items():
default = (
...
if param.default is inspect.Parameter.empty
else param.default
)
annotation = (
param.annotation
if param.annotation is not inspect.Parameter.empty
else Any
)
fields[param_name] = (annotation, default)
model = create_model(f"{tool_name}_Input", **fields)
# Store the schema and the bound executable function
self._tools[tool_name] = {
"name": tool_name,
"description": description,
"model": model,
"func": func,
"schema": {
"type": "function",
"function": {
"name": tool_name,
"description": description,
"parameters": model.model_json_schema(),
},
},
}
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator
def get_tool_schemas(self) -> list[dict]:
"""Returns schemas formatted for LLM tool calling APIs."""
return [tool["schema"] for tool in self._tools.values()]
def dispatch(self, tool_name: str, raw_arguments: str | dict) -> Any:
"""Validates payload against bound schema and executes the target function."""
if tool_name not in self._tools:
raise KeyError(f"Tool '{tool_name}' not found in registry.")
tool = self._tools[tool_name]
# Parse string arguments if received from JSON-based LLM response
if isinstance(raw_arguments, str):
raw_arguments = json.loads(raw_arguments)
# Validate arguments through the bound Pydantic model
validated_args = tool["model"](**raw_arguments)
# Execute target function using validated values
return tool["func"](**validated_args.model_dump())
How to Use the Pattern
1. Define Tools with the Decorator
You can either pass an explicit Pydantic class or let the decorator infer fields from native type annotations:
registry = ToolRegistry()
# Approach A: Explicit Pydantic input schema
class WeatherInput(BaseModel):
city: str = Field(..., description="Target city name")
unit: str = Field("celsius", description="'celsius' or 'fahrenheit'")
@registry.register(schema_model=WeatherInput)
def get_current_weather(city: str, unit: str = "celsius") -> str:
"""Fetch weather information for a specific location."""
return f"Weather in {city}: 21° {unit.upper()}, clear skies."
# Approach B: Native type hints auto-inferred into a Pydantic model
@registry.register()
def calculate_power(base: float, exponent: float = 2.0) -> float:
"""Compute base raised to the power of exponent."""
return base**exponent
2. Inspecting the Generated Schemas
Calling registry.get_tool_schemas() exports the tool specifications directly formatted for provider APIs:
print(json.dumps(registry.get_tool_schemas(), indent=2))
3. Handling Incoming LLM Tool Calls via Dispatch
When the model issues a tool call, routing requires a single line without explicit conditional branching:
# Simulated tool call payload returned by the LLM
tool_call_name = "get_current_weather"
raw_payload = '{"city": "Austin", "unit": "fahrenheit"}'
# Dispatch automatically validates arguments and runs the bound function
result = registry.dispatch(tool_call_name, raw_payload)
print(result)
# Output: Weather in Austin: 21° FAHRENHEIT, clear skies.
Key Advantages of this Pattern
Single Source of Truth: Function logic, input contracts, and metadata are co-located in the function definition rather than split across multiple mapping dictionaries.
Fail-Fast Validation: Malformed or hallucinated parameters are caught by Pydantic before reaching execution code, returning descriptive validation errors to the model.
Zero Dispatch Boilerplate: Adding or removing tools requires writing or deleting a decorated function; no routing switches or dispatch maps need manual synchronization.
Provide a Python application using Google Gemini API to create AI agents from a list of pairs: [[skill.md and pydantic_class.py], ...] supporting decorator-based tool dispatch, with binding. Results of all agent activity should be saved as markdown files
The complete application below uses the official Google Gen AI SDK (google-genai). It accepts pairs of [skill.md, pydantic_class.py], dynamically loads schemas, binds executable functions via decorators, executes tool calls in an agentic loop, and logs each agent run's thoughts, tool interactions, and final answers to Markdown files.
Prerequisites
Install the official Google Gen AI SDK and Pydantic:
pip install google-genai pydantic
Set your API key:
export GEMINI_API_KEY="your-gemini-api-key"
Project Layout
agent_system/
├── agent_orchestrator.py
├── run_agents.py
├── skills/
│ ├── researcher_skill.md
│ ├── researcher_tools.py
│ ├── math_skill.md
│ └── math_tools.py
└── logs/
1. agent_orchestrator.py
This module defines:
The
ToolRegistryproviding decorator-based dispatch with automatic Pydantic validation and context binding.The
GeminiAgentrunning the tool loop and serializing execution history directly into.mddocuments.
import datetime
import importlib.util
import inspect
import json
import os
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Type
from google import genai
from google.genai import types
from pydantic import BaseModel
def clean_schema(d: Any) -> Any:
"""Removes JSON-schema keys not accepted by Gemini API's OpenAPI specification."""
if isinstance(d, dict):
# Gemini expects standard OpenAPI fields; remove JSON schema dialect markers
cleaned = {
k: clean_schema(v)
for k, v in d.items()
if k not in ("title", "$schema", "additionalProperties")
}
return cleaned
elif isinstance(d, list):
return [clean_schema(i) for i in d]
return d
class ToolRegistry:
"""Provides decorator-based tool dispatch with Pydantic validation."""
def __init__(self):
self._registry: Dict[str, Dict[str, Any]] = {}
def tool(self, schema_model: Type[BaseModel]):
"""Decorator to register a function and bind it to a Pydantic input schema."""
def decorator(func: Callable[..., Any]):
name = schema_model.__name__
desc = (
schema_model.__doc__
or func.__doc__
or f"Executes {func.__name__}"
)
raw_schema = schema_model.model_json_schema()
self._registry[name] = {
"name": name,
"description": desc.strip(),
"model": schema_model,
"func": func,
"schema": clean_schema(raw_schema),
}
return func
return decorator
def build_gemini_tools(self) -> List[types.Tool]:
"""Converts registered tools into Gemini types.Tool declarations."""
declarations = []
for name, meta in self._registry.items():
fn_decl = types.FunctionDeclaration(
name=meta["name"],
description=meta["description"],
parameters=meta["schema"],
)
declarations.append(fn_decl)
return (
[types.Tool(function_declarations=declarations)]
if declarations
else []
)
def dispatch(self, tool_name: str, arguments: dict) -> Any:
"""Validates payload against bound Pydantic model and executes the function."""
if tool_name not in self._registry:
raise KeyError(f"Tool '{tool_name}' is not registered.")
target = self._registry[tool_name]
model_cls: Type[BaseModel] = target["model"]
handler: Callable = target["func"]
# Validate arguments using the bound Pydantic model
validated_input = model_cls(**arguments)
# Execute bound function with kwargs from validated model
return handler(**validated_input.model_dump())
class GeminiAgent:
"""Agent dynamically configured from skill.md and pydantic_class.py."""
def __init__(
self,
skill_path: str | Path,
pydantic_code_path: str | Path,
model_name: str = "gemini-2.5-flash",
output_dir: str = "./logs",
):
self.skill_path = Path(skill_path)
self.pydantic_code_path = Path(pydantic_code_path)
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.model_name = model_name
self.client = genai.Client()
self.system_instruction = self._load_skill()
self.registry = ToolRegistry()
self._load_and_bind_tools()
self.tools = self.registry.build_gemini_tools()
def _load_skill(self) -> str:
if not self.skill_path.is_file():
raise FileNotFoundError(f"Missing skill file: {self.skill_path}")
return self.skill_path.read_text(encoding="utf-8")
def _load_and_bind_tools(self):
"""Dynamically imports the python file and executes registration decorators."""
if not self.pydantic_code_path.is_file():
raise FileNotFoundError(
f"Missing python file: {self.pydantic_code_path}"
)
module_name = f"dynamic_module_{self.pydantic_code_path.stem}"
spec = importlib.util.spec_from_file_location(
module_name, self.pydantic_code_path
)
if spec is None or spec.loader is None:
raise ImportError(
f"Could not load module from {self.pydantic_code_path}"
)
module = importlib.util.module_from_spec(spec)
# Inject the registry decorator directly into the module's namespace
# so functions in the file can decorate themselves using @tool(...)
module.tool = self.registry.tool
spec.loader.exec_module(module)
def run(self, user_prompt: str, max_iterations: int = 6) -> str:
"""Runs the agentic execution loop and writes results to Markdown."""
session_id = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
log_name = f"{self.skill_path.stem}_{session_id}.md"
log_file = self.output_dir / log_name
# Initialize markdown logging buffer
md_log = [
f"# Agent Execution Log: {self.skill_path.stem}",
f"- **Date/Time**: {datetime.datetime.now().isoformat()}",
f"- **Skill File**: `{self.skill_path}`",
f"- **Tool Module**: `{self.pydantic_code_path}`",
f"- **Model**: `{self.model_name}`",
"\n## System Instruction (Skill)\n",
"```markdown",
self.system_instruction.strip(),
"```\n",
"## User Query\n",
f"> {user_prompt}\n",
"## Activity Trail\n",
]
contents: List[types.Content] = [
types.Content(
role="user", parts=[types.Part.from_text(text=user_prompt)]
)
]
iteration = 0
final_answer = ""
while iteration < max_iterations:
iteration += 1
config = types.GenerateContentConfig(
system_instruction=self.system_instruction,
tools=self.tools if self.tools else None,
temperature=0.0,
)
response = self.client.models.generate_content(
model=self.model_name, contents=contents, config=config
)
# Record model content back into history
model_parts = response.candidates[0].content.parts
contents.append(
types.Content(role="model", parts=list(model_parts))
)
# Check for function calls
function_calls = [
part.function_call
for part in model_parts
if part.function_call is not None
]
if not function_calls:
final_answer = response.text or ""
md_log.append(
f"### Turn {iteration}: Final Response\n\n{final_answer}\n"
)
break
tool_response_parts = []
for call in function_calls:
fn_name = call.name
fn_args = dict(call.args) if call.args else {}
md_log.append(f"### Turn {iteration}: Tool Invocation")
md_log.append(f"- **Tool Called**: `{fn_name}`")
md_log.append(
f"- **Arguments**:\n```json\n{json.dumps(fn_args, indent=2)}\n```"
)
try:
result = self.registry.dispatch(fn_name, fn_args)
formatted_result = (
result
if isinstance(result, (dict, list))
else {"output": str(result)}
)
except Exception as exc:
formatted_result = {"error": str(exc)}
md_log.append(
f"- **Result**:\n```json\n{json.dumps(formatted_result, indent=2)}\n```\n"
)
tool_response_parts.append(
types.Part.from_function_response(
name=fn_name, response=formatted_result
)
)
contents.append(
types.Content(role="tool", parts=tool_response_parts)
)
md_log.append("## Final Output Delivered to User\n")
md_log.append(final_answer if final_answer else "_No final output generated._")
# Save markdown log
log_file.write_text("\n".join(md_log), encoding="utf-8")
print(f"Log written to: {log_file}")
return final_answer
2. Example Skill & Tool Files
Pair 1: Research Agent
skills/researcher_skill.md:
# Role: Technology Analyst
You analyze technologies by retrieving verified specifications and summarizing the pros and cons concisely.
skills/researcher_tools.py:
from pydantic import BaseModel, Field
# 1. Define Pydantic Class (Schema)
class SearchTechDatabase(BaseModel):
"""Searches technical specifications and release dates for hardware."""
query: str = Field(..., description="Device or GPU model name to lookup")
spec_field: str = Field(
"all", description="Specific property: 'memory', 'release_date', or 'all'"
)
# 2. Decorator binds the schema model to the handler
@tool(SearchTechDatabase)
def handle_search(query: str, spec_field: str = "all") -> dict:
# Simulated database lookup
specs_mock = {
"Nvidia RTX 5090": {
"memory": "32GB GDDR7",
"release_date": "2025",
"bus": "512-bit",
},
"Apple M4 Max": {
"memory": "Up to 128GB Unified",
"release_date": "Late 2024",
"cores": "16 CPU / 40 GPU",
},
}
for item, data in specs_mock.items():
if query.lower() in item.lower():
if spec_field != "all" and spec_field in data:
return {item: {spec_field: data[spec_field]}}
return {item: data}
return {"message": f"No specifications found for '{query}'"}
Pair 2: Scientific Compute Agent
skills/math_skill.md:
# Role: Exact Computational Engine
You never approximate mathematical results. You always perform computations using the provided calculation tools.
skills/math_tools.py:
import math
from pydantic import BaseModel, Field
class FactorialCalculator(BaseModel):
"""Computes the exact mathematical factorial of an integer n."""
n: int = Field(..., ge=0, description="Non-negative integer")
@tool(FactorialCalculator)
def handle_factorial(n: int) -> dict:
return {"input": n, "factorial": math.factorial(n)}
3. run_agents.py
Instantiates agents dynamically from an arbitrary [[skill, pydantic_tools], ...] list:
from agent_orchestrator import GeminiAgent
AGENT_PAIRS = [
[
"skills/researcher_skill.md",
"skills/researcher_tools.py",
"What are the memory specs and release date for the Nvidia RTX 5090?",
],
[
"skills/math_skill.md",
"skills/math_tools.py",
"What is the factorial of 12?",
],
]
def main():
for skill_file, tool_file, query in AGENT_PAIRS:
print(f"\n{'='*60}")
print(f"Spinning up agent: [{skill_file}] + [{tool_file}]")
print(f"Task: {query}")
print(f"{'='*60}")
agent = GeminiAgent(
skill_path=skill_file,
pydantic_code_path=tool_file,
model_name="gemini-2.5-flash",
output_dir="./logs",
)
response = agent.run(query)
print(f"\nFinal Response:\n{response}\n")
if __name__ == "__main__":
main()
Generated Markdown Artifact Example
For each run, a timestamped file like ./logs/researcher_skill_20261029_002200.md is generated:
# Agent Execution Log: researcher_skill
- **Date/Time**: 2026-10-29T00:22:00.123456
- **Skill File**: `skills/researcher_skill.md`
- **Tool Module**: `skills/researcher_tools.py`
- **Model**: `gemini-2.5-flash`
## System Instruction (Skill)
```markdown
# Role: Technology Analyst
You analyze technologies by retrieving verified specifications and summarizing the pros and cons concisely.
User Query
What are the memory specs and release date for the Nvidia RTX 5090?
Activity Trail
Turn 1: Tool Invocation
Tool Called:
SearchTechDatabaseArguments:
{
"query": "Nvidia RTX 5090",
"spec_field": "all"
}
Result:
{
"Nvidia RTX 5090": {
"memory": "32GB GDDR7",
"release_date": "2025",
"bus": "512-bit"
}
}
Turn 2: Final Response
The Nvidia RTX 5090 features 32GB of GDDR7 memory on a 512-bit memory bus and was released in 2025.
From <https://gemini.google.com/app/845e519f1071c46e> Google Gemini (3.8 Flash)
Comments
Post a Comment