SAKURA Internetは無料で3000リクエストも使えて太っ腹です。
夏休みの自由研究のため、AIエージェント作成の解説の記録！
400くらいリクエストを使っていました。
残りリクエストも夏休みも少なくこれが集大成。これで終わります。
結局3000リクエストは使いきれず余りました。
業務利用でなければ十分な量でした。
可能なら今後も継続して使いたいです！！！
AIエージェントを自作して、やっと「ループエンジニアリング」の意味がわかった
そりゃ、ループエンジニアリングの言い出しっぺのBoris Chernyさんも乗っかったPeter Steinbergerさんもエージェントの開発者。
この視点で語られることをイチアプリ作成で理解するのは無理かあったのだと。
そう気が付きました。
前回LLMチャット作ったので、AIエージェントを自作してみた。
前回、今回どちらも簡易であるわけだけど。(例によって動くコードは後半に)
最初は、
「LLM APIを呼んで、ツールを実行させればいいんでしょ？」
くらいに考えていた。
実際に作ってみる。
すると、思ったより普通に動く。
ファイルを読む。
ファイルを探す。
コードを書く。
コードを編集する。
コマンドを実行する。
そして、結果をまたLLMに渡す。
……あれ？
これ、Claude CodeやCodexの中で起きていることと、かなり近い。
AIエージェントの本体は、LLMそのものではない。
むしろ重要なのは、LLMをどう回すか。回し続けるかだ。
自作してみたら「ループ」が見えた
今回作ったエージェントは、かなりシンプル。
構成はこんな感じ。
ユーザー
↓
System Prompt
↓
Context
↓
LLM
↓
Tool Call
↓
Permission
↓
Tool実行
↓
Tool Result
↓
Contextへ追加
↓
LLM
↓
Tool Call
↓
……
つまり、
LLM → ツール → 結果 → LLM → ツール → 結果……
というループ。
実装がわかりやすい。（2重ループの箇所）
while self.turn_count < self.max_turns:
self.turn_count += 1
messages = self.context.build_messages()
tools = self.context.get_tool_definitions()
response = self.llm.chat(
messages,
tools=tools
)
if response.has_tool_calls():
for tc in response.tool_calls:
result = self.tools.execute(
tc.name,
tc.arguments
)
self.context.add_tool_result(
tc.id,
tc.name,
result
)
else:
final_response = response.content
break
これだけでも、AIエージェントの核心がかなり見える。
AIに一回質問して終わり、ではない。
AIが考える。
↓
ツールを使う。
↓
結果を見る。
↓
もう一度考える。
↓
またツールを使う。
↓
結果を見る。
↓
目的を達成するまで続ける。
この「回し方」がエージェントだった。
そして「4つのエンジニアリング」がつながった
ここで、やっと腑に落ちた。
- プロンプトエンジニアリング
- コンテキストエンジニアリング
- ハーネスエンジニアリング
- ループエンジニアリング
AIエージェントを作るための4つの概念でしょ
① プロンプトエンジニアリング
まず、AIに何をさせるか。
今回のエージェントにもSystem Promptがある。
You are MyAgent, an autonomous coding assistant.
You help users by reading code,
writing code,
editing files,
running commands,
and fixing bugs.
AIに「あなたは何者なのか」「何をするのか」を与える。
これが最初の入口。
② コンテキストエンジニアリング
次に重要なのが、
何をAIに見せるか。
今回の実装ではContextManagerを作った。
System Prompt。
ユーザーの入力。
AIの回答。
Tool Call。
Tool Result。
これらを履歴として保持して、次のLLM呼び出しに渡す。
def build_messages(self):
result = [
{
"role": "system",
"content": self.system_prompt
}
]
result.extend(self.messages)
return result
AIエージェントでは、①プロンプトと②コンテキストが
「①どんなプロンプトを書くか」 と「②次のAIに、何を記憶として渡すか」
として実装された。
③ ハーネスエンジニアリング
AIに自由にやらせるのは危ない。
例えば、rm -rf / なんて実行されたら終わる。
そこで、AIとツールの間にガードを入れる。
DANGEROUS_PATTERNS = [
"rm -rf /",
"rm -rf ~",
"rm -rf *",
"mkfs",
":(){ :|:& };:",
]
のような危険コマンドをブラックリストでブロック。
read_file → auto
list_files → auto
search_files → auto
write_file → ask
edit_file → ask
run_command → ask
のように、ツールごとに権限も設定できる。
AIに何をさせるかだけではなく、
AIが何をしてはいけないか
これがハーネス。
④ そしてループエンジニアリング
ここが今回、一番大きかった発見。
AIエージェントは、
Prompt → LLM → Answer
ではない。
Prompt → LLM → Tool → Result→Context → LLM → Tool → Result → Context → LLM → …
AIの能力を「1回の回答」ではなく「何回回せるか」で引き出す。
今回のMyAgentでは最大ターン数も設定している。
max_turns = 50
エージェントを回す。
while self.turn_count < self.max_turns:
できるだけ少ないループで回数を減らせるのがベスト。ここが設計の要でしょうね。
悪いループは、何やってんだAIになる
LLM → 間違える → LLM → さらに間違える → LLM → もっと間違える
対策は ループ × コンテキスト × ツール × ガードレールの設計が必要
AIエージェントを作ると、LLMの見え方が変わる
ChatGPTのようなチャットだけを使っていると、
「AIが回答している」
ように見える。
でもエージェントを作ってみると、景色が変わる。
実際に起きていることは、
┌──────────────┐
│ LLM │
└──────┬───────┘
│
Tool Call
↓
┌──────────────┐
│ Tool │
└──────┬───────┘
│
Tool Result
↓
┌──────────────┐
│ Context │
└──────┬───────┘
│
└────────→ LLM
これを延々と回している。
だから、
「AIエージェントを作る」＝「LLMを賢くする」
ではない。
「LLMをどういう環境に置いて、どう回すかを設計する」
という話だった。
Claude CodeやCodexがすごい理由も少し見えてきた
ここまで自作してみると、
Claude CodeやCodexのようなCoding Agentが単なる「LLMチャット」ではないことがよくわかる。
重要なのは、
- コンテキスト管理
- ツール定義
- ツール実行
- 権限制御
- エラー処理
- プロンプト
- 状態管理
- ループ制御
- 最大ターン数
- ログ
- ユーザー確認
などを全部まとめて設計していること。
今回の自作エージェントでも、実際に、
ReadFileTool
WriteFileTool
EditFileTool
ListFilesTool
SearchFilesTool
RunCommandTool
というツールを登録している。
そしてLLMが必要に応じて呼び出す。
Coding Agentは「制御システム + LLM + ツール + ループ 」が肝で
制御システムがコンテキストとハーネス。ループはループ。
結局、AIエージェント開発で重要なのは何なのか
今回、自作してみて整理するとこうなった。
| レイヤー | やること |
|---|---|
| プロンプト | AIに何をさせるか |
| コンテキスト | AIに何を見せるか |
| ハーネス | AIをどう安全に制御するか |
| ツール | AIに何ができるようにするか |
| ループ | AIをどう繰り返し動かすか |
そして全部をつなぐと、
┌─────────────┐
│ Prompt │
└──────┬──────┘
↓
┌─────────────┐
│ Context │
└──────┬──────┘
↓
┌─────────────┐
│ LLM │
└──────┬──────┘
↓
┌─────────────┐
│ Tool │
└──────┬──────┘
↓
┌─────────────┐
│ Harness │
└──────┬──────┘
↓
┌─────────────┐
│ Result │
└──────┬──────┘
│
└────────→ Context
│
└──→ LLM
という循環になる。
これが、今回自作して一番腹落ちした部分。
「エージェントを作る」とは何だったのか
最初は、
「PythonでLLM APIを呼んで、ファイル操作できるようにする」
くらいのつもりだった。
でも実際に作ってみたら違った。
AIに能力を与えることではなく、AIが能力を使い続けられる仕組みを作ること。
AIが暴走しないように囲いを作ること。
必要な情報を次のループへ渡すこと。
その全部を設計する。
プロンプトエンジニアリング
コンテキストエンジニアリング
ハーネスエンジニアリング
ループエンジニアリング
という言葉が、やっと一つにつながった。
AIエージェント開発とは、LLMを中心にした「制御ループ」を設計すること。
ここまでわかると、Claude CodeやCodexを見る目も変わる。
「このループ、どう設計してるんだ？」になる。
だから、ここが妙に腹落ち。
ループエンジニアリングの言い出しっぺのBoris Chernyさんも
乗っかったPeter Steinbergerさんもエージェントの開発者。
ここで満足し終了した。
付録 ソースコード全文
クリックで展開
結構ボリューム大きいですが、Linuxで動作確認済み。
#初期化
Myagent.py --init
#設定ファイルにAPIキーを記載
vi config.json
#起動
Myagent.py
Myagent.py
#!/usr/bin/python3
# Module: myagent.context
from typing import Any, Dict, List, Optional
class ContextManager:
def __init__(self, system_prompt: str, max_history: int = 100):
self.system_prompt = system_prompt
self.max_history = max_history
self.messages: List[Dict[str, Any]] = []
self.tool_definitions: List[Dict[str, Any]] = []
self._initialized = False
def set_tool_definitions(self, tools: List[Dict[str, Any]]) -> None:
self.tool_definitions = tools
def build_messages(self) -> List[Dict[str, Any]]:
result = [{"role": "system", "content": self.system_prompt}]
result.extend(self.messages)
return result
def add_user_message(self, content: str) -> None:
self.messages.append({"role": "user", "content": content})
self._trim_history()
def add_assistant_message(
self, content: Optional[str] = None, tool_calls: Optional[List[Dict]] = None
) -> None:
msg: Dict[str, Any] = {"role": "assistant"}
if content is not None:
msg["content"] = content
if tool_calls:
msg["tool_calls"] = tool_calls
self.messages.append(msg)
self._trim_history()
def add_tool_result(self, tool_call_id: str, name: str, result: str) -> None:
self.messages.append(
{
"role": "tool",
"tool_call_id": tool_call_id,
"name": name,
"content": str(result),
}
)
self._trim_history()
def get_tool_definitions(self) -> List[Dict[str, Any]]:
return self.tool_definitions
def get_messages(self) -> List[Dict[str, Any]]:
return list(self.messages)
def clear(self) -> None:
self.messages = []
def _trim_history(self) -> None:
if len(self.messages) > self.max_history:
excess = len(self.messages) - self.max_history
self.messages = self.messages[excess:]
def estimate_tokens(self) -> int:
total = 0
for msg in self.messages:
content = msg.get("content", "")
if content:
total += len(content) // 4
return total
# Module: myagent.llm
import json
import logging
import os
from abc import ABC, abstractmethod
from datetime import datetime
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
import requests
logger = logging.getLogger("myagent.llm")
@dataclass
class ToolCall:
id: str
name: str
arguments: Dict[str, Any]
@dataclass
class LLMResponse:
content: Optional[str] = None
tool_calls: List[ToolCall] = field(default_factory=list)
raw_response: Optional[Dict] = None
usage: Optional[Dict[str, int]] = None
def has_tool_calls(self) -> bool:
return len(self.tool_calls) > 0
class LLMProvider(ABC):
@abstractmethod
def chat(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> LLMResponse:
pass
class OpenAICompatibleProvider(LLMProvider):
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = "https://api.openai.com/v1",
default_model: str = "gpt-4o-mini",
timeout: int = 120,
):
self.api_key = api_key or os.environ.get("OPENAI_API_KEY", "")
self.base_url = base_url.rstrip("/")
self.default_model = default_model
self.timeout = timeout
if not self.api_key:
raise ValueError(
"API key is required. Set OPENAI_API_KEY env var or pass api_key."
)
masked_key = self.api_key[:8] + "..." + self.api_key[-4:] if len(self.api_key) > 12 else "..."
logger.info(
"Provider initialized: base_url=%s, model=%s",
self.base_url,
self.default_model,
)
logger.debug("API key: %s", masked_key)
def _prepare_tools(self, tools: List[Dict]) -> List[Dict]:
return [
{
"type": "function",
"function": {
"name": t["name"],
"description": t["description"],
"parameters": t.get("parameters", {"type": "object", "properties": {}}),
},
}
for t in tools
]
def chat(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> LLMResponse:
payload: Dict[str, Any] = {
"model": model or self.default_model,
"messages": messages,
}
if tools:
payload["tools"] = self._prepare_tools(tools)
payload["tool_choice"] = "auto"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
url = f"{self.base_url}/chat/completions"
logger.debug(
"LLM request: model=%s, messages=%d, tools=%s",
payload["model"],
len(payload["messages"]),
"yes" if tools else "no",
)
start_time = datetime.now()
try:
response = requests.post(
url, headers=headers, json=payload, timeout=self.timeout
)
response.raise_for_status()
data = response.json()
except requests.exceptions.HTTPError as e:
logger.error("HTTP error response: %s", e.response.text)
raise
except requests.exceptions.RequestException as e:
logger.error("Request failed: %s", e)
raise
elapsed_ms = (datetime.now() - start_time).total_seconds() * 1000
logger.info("LLM response received in %.1f ms", elapsed_ms)
choice = data["choices"][0]
message = choice["message"]
content = message.get("content")
tool_calls = []
for tc in message.get("tool_calls", []):
if tc["type"] == "function":
func = tc["function"]
tool_calls.append(
ToolCall(
id=tc["id"],
name=func["name"],
arguments=json.loads(func["arguments"]),
)
)
usage = data.get("usage")
if usage:
logger.info(
"Tokens: prompt=%d, completion=%d, total=%d",
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0),
usage.get("total_tokens", 0),
)
if content:
logger.debug("Response content: %s", content[:500])
if tool_calls:
for tc in tool_calls:
logger.debug("Tool call: %s(%s)", tc.name, json.dumps(tc.arguments, ensure_ascii=False))
return LLMResponse(
content=content,
tool_calls=tool_calls,
raw_response=data,
usage=usage,
)
class DebugProvider(LLMProvider):
def __init__(self, default_model: str = "debug"):
self.default_model = default_model
def chat(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict]] = None,
model: Optional[str] = None,
) -> LLMResponse:
logger.info("DEBUG mode: skipping API call")
logger.debug("Messages: %d, Tools: %s", len(messages), "yes" if tools else "no")
if tools:
tool_names = [t["name"] for t in tools]
logger.debug("Available tools: %s", ", ".join(tool_names))
for msg in reversed(messages):
role = msg.get("role", "")
if role in ("user", "assistant"):
content = msg.get("content", "") or ""
logger.debug("Last %s message: %s", role, content[:200])
break
return LLMResponse(
content="DEBUG: This is a test response. No API call was made.",
usage={"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
)
def create_provider(config: Dict[str, Any]) -> LLMProvider:
api_key = config.get("api_key", "")
if api_key == "DEBUG":
logger.info("Using DebugProvider (dry-run, no API calls)")
return DebugProvider(default_model=config.get("model", "debug"))
provider_type = config.get("provider", "openai")
if provider_type in ("openai", "sakura"):
return OpenAICompatibleProvider(
api_key=api_key,
base_url=config.get("base_url", "https://api.openai.com/v1"),
default_model=config.get("model", "gpt-4o-mini"),
timeout=config.get("timeout", 120),
)
raise ValueError(f"Unknown provider type: {provider_type}")
# Module: myagent.permission
from typing import Dict
class PermissionManager:
DEFAULT_LEVELS = {
"read_file": "auto",
"list_files": "auto",
"search_files": "auto",
"write_file": "ask",
"edit_file": "ask",
"run_command": "ask",
}
DANGEROUS_PATTERNS = [
"rm -rf /",
"rm -rf ~",
"rm -rf *",
"mkfs",
":(){ :|:& };:",
]
def __init__(self, overrides: Dict[str, str] = None):
self.levels = dict(self.DEFAULT_LEVELS)
if overrides:
self.levels.update(overrides)
def check(self, tool_name: str, arguments: Dict = None) -> bool:
level = self.levels.get(tool_name, "ask")
if tool_name == "run_command" and arguments:
cmd = arguments.get("command", "")
for pattern in self.DANGEROUS_PATTERNS:
if pattern in cmd:
print(f"[Permission] BLOCKED dangerous command: {cmd}")
return False
if level == "auto":
return True
if level == "deny":
print(f"[Permission] Tool '{tool_name}' is denied by policy.")
return False
arg_summary = ""
if arguments:
if tool_name in ("write_file", "edit_file"):
arg_summary = f" path={arguments.get('path', '?')}"
elif tool_name == "run_command":
arg_summary = f" command={arguments.get('command', '?')[:60]}"
prompt = f"Allow {tool_name}{arg_summary}? [y/N/s(always Skip)/a(always Allow)] "
response = input(prompt).strip().lower()
if response in ("a", "always"):
self.levels[tool_name] = "auto"
return True
if response in ("s", "skip"):
self.levels[tool_name] = "deny"
return False
return response in ("y", "yes")
def set_level(self, tool_name: str, level: str) -> None:
if level not in ("auto", "ask", "deny"):
raise ValueError("Level must be auto, ask, or deny")
self.levels[tool_name] = level
def get_summary(self) -> str:
lines = ["Permission levels:"]
for tool, level in sorted(self.levels.items()):
lines.append(f" {tool}: {level}")
return "\n".join(lines)
# Module: myagent.prompt_loader
import json
import os
from pathlib import Path
from typing import Dict, List, Optional
class PromptLoader:
DEFAULT_PROMPT = (
"You are MyAgent, an autonomous coding assistant. "
"You help users by reading code, writing code, editing files, "
"running commands, and fixing bugs."
)
def __init__(self, prompts_dir: str):
self.prompts_dir = Path(prompts_dir)
self.seq_file = self.prompts_dir / "00-seq.txt"
def _read_sequence(self) -> List[str]:
if not self.seq_file.exists():
return []
ids = []
with open(self.seq_file, "r", encoding="utf-8") as f:
for line in f:
line = line.split("#")[0].strip()
if line:
ids.append(line)
return ids
def _load_prompt_file(self, prompt_id: str) -> Optional[str]:
file_path = self.prompts_dir / f"{prompt_id}.txt"
if not file_path.exists():
print(f"[PromptLoader] Warning: {file_path} not found, skipping")
return None
with open(file_path, "r", encoding="utf-8") as f:
return f.read().strip()
def load(self, separator: str = "\n\n---\n\n") -> str:
if "_BUNDLED_PROMPTS" in globals() and "_BUNDLED_SEQ" in globals():
bp = globals()["_BUNDLED_PROMPTS"]
bs = globals()["_BUNDLED_SEQ"]
parts = []
for pid in bs:
if pid in bp:
parts.append(bp[pid])
if parts:
return separator.join(parts)
return self.DEFAULT_PROMPT
prompt_ids = self._read_sequence()
if not prompt_ids:
print("[PromptLoader] No sequence file found, using default prompt")
return self.DEFAULT_PROMPT
parts = []
for pid in prompt_ids:
content = self._load_prompt_file(pid)
if content is not None:
parts.append(content)
if not parts:
print("[PromptLoader] No prompt files loaded, using default prompt")
return self.DEFAULT_PROMPT
return separator.join(parts)
# Module: myagent.tools.base
from abc import ABC, abstractmethod
from typing import Any, Dict
class Tool(ABC):
@property
@abstractmethod
def name(self) -> str:
pass
@property
@abstractmethod
def description(self) -> str:
pass
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {},
}
@abstractmethod
def execute(self, arguments: Dict[str, Any]) -> str:
pass
def to_definition(self) -> Dict[str, Any]:
return {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}
def _require(self, arguments: Dict[str, Any], key: str) -> Any:
if key not in arguments:
raise ValueError(f"Missing required argument: '{key}' for tool '{self.name}'")
return arguments[key]
# Module: myagent.logger
import logging
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional
DEFAULT_LOG_DIR = "logs"
_logger_cache: dict[str, logging.Logger] = {}
def get_logger(name: str) -> logging.Logger:
if name in _logger_cache:
return _logger_cache[name]
logger = logging.getLogger(name)
_logger_cache[name] = logger
return logger
def setup_logging(
log_dir: Optional[str] = None,
console_level: int = logging.INFO,
file_level: int = logging.DEBUG,
) -> Path:
root_logger = logging.getLogger("myagent")
root_logger.setLevel(logging.DEBUG)
for handler in list(root_logger.handlers):
root_logger.removeHandler(handler)
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(console_level)
console_format = logging.Formatter(
"%(asctime)s [%(levelname)s] %(message)s",
datefmt="%H:%M:%S",
)
console_handler.setFormatter(console_format)
root_logger.addHandler(console_handler)
log_dir_path = Path(log_dir or DEFAULT_LOG_DIR)
log_dir_path.mkdir(parents=True, exist_ok=True)
date_str = datetime.now().strftime("%Y-%m-%d")
log_file = log_dir_path / f"myagent_{date_str}.log"
file_handler = logging.FileHandler(str(log_file), encoding="utf-8")
file_handler.setLevel(file_level)
file_format = logging.Formatter(
"%(asctime)s [%(levelname)s] %(name)s:%(lineno)d - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
file_handler.setFormatter(file_format)
root_logger.addHandler(file_handler)
root_logger.info("Logging initialized. File: %s", log_file)
return log_file
# Module: myagent.tools.read_file
from typing import Any, Dict
class ReadFileTool(Tool):
@property
def name(self) -> str:
return "read_file"
@property
def description(self) -> str:
return "Read the contents of a file at the given path."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute or relative path to the file",
},
"offset": {
"type": "integer",
"description": "Line number to start reading from (1-indexed)",
},
"limit": {
"type": "integer",
"description": "Maximum number of lines to read",
},
},
"required": ["path"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
path = self._require(arguments, "path")
offset = arguments.get("offset", 1)
limit = arguments.get("limit")
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
start = max(0, offset - 1)
end = len(lines)
if limit is not None:
end = start + limit
selected = lines[start:end]
content = "".join(selected)
numbered_lines = []
for i, line in enumerate(selected, start=start + 1):
numbered_lines.append(f"{i:4d}: {line}")
result = "".join(numbered_lines)
if limit and end < len(lines):
result += f"\n... ({len(lines) - end} more lines)"
return result if result else "(empty file)"
except FileNotFoundError:
return f"ERROR: File not found: {path}"
except PermissionError:
return f"ERROR: Permission denied: {path}"
except Exception as e:
return f"ERROR: {e}"
# Module: myagent.tools.run_command
import os
import subprocess
import tempfile
from typing import Any, Dict
class RunCommandTool(Tool):
@property
def name(self) -> str:
return "run_command"
@property
def description(self) -> str:
return "Execute a shell command in the current working directory. Use this to run tests, check syntax, or use git."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "Shell command to execute",
},
"timeout": {
"type": "integer",
"description": "Timeout in seconds (default: 60)",
},
"cwd": {
"type": "string",
"description": "Working directory for the command (default: current directory)",
},
},
"required": ["command"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
command = self._require(arguments, "command")
timeout = arguments.get("timeout", 60)
try:
cwd = arguments.get("cwd", os.getcwd())
except FileNotFoundError:
cwd = arguments.get("cwd", "/tmp")
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
cwd=cwd,
)
output_lines = []
if result.stdout:
output_lines.append("STDOUT:")
output_lines.append(result.stdout.rstrip())
if result.stderr:
output_lines.append("STDERR:")
output_lines.append(result.stderr.rstrip())
output_lines.append(f"\nExit code: {result.returncode}")
return "\n".join(output_lines)
except subprocess.TimeoutExpired:
return f"ERROR: Command timed out after {timeout} seconds"
except Exception as e:
return f"ERROR: {type(e).__name__}: {e}"
# Module: myagent.tools.list_files
import os
from typing import Any, Dict
class ListFilesTool(Tool):
@property
def name(self) -> str:
return "list_files"
@property
def description(self) -> str:
return "List files and directories at a given path. Can list recursively."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Directory path to list (default: current directory)",
},
"recursive": {
"type": "boolean",
"description": "List recursively (default: false)",
},
"max_depth": {
"type": "integer",
"description": "Maximum recursion depth (default: 3 for recursive)",
},
},
}
def execute(self, arguments: Dict[str, Any]) -> str:
path = arguments.get("path", ".")
recursive = arguments.get("recursive", False)
max_depth = arguments.get("max_depth", 3)
try:
target = os.path.abspath(path)
if not os.path.exists(target):
return f"ERROR: Path not found: {target}"
if not os.path.isdir(target):
return f"ERROR: Not a directory: {target}"
lines = []
self._list_dir(target, "", lines, recursive, 0, max_depth)
return "\n".join(lines) if lines else "(empty directory)"
except PermissionError:
return f"ERROR: Permission denied: {path}"
except Exception as e:
return f"ERROR: {e}"
def _list_dir(self, root: str, prefix: str, lines: list, recursive: bool, depth: int, max_depth: int) -> None:
try:
entries = sorted(os.listdir(root))
except PermissionError:
lines.append(f"{prefix}[permission denied]")
return
for i, entry in enumerate(entries):
entry_path = os.path.join(root, entry)
is_last = i == len(entries) - 1
connector = "└── " if is_last else "├── "
indent = " " if is_last else "│ "
if os.path.isdir(entry_path):
lines.append(f"{prefix}{connector}{entry}/")
if recursive and depth < max_depth:
self._list_dir(entry_path, prefix + indent, lines, recursive, depth + 1, max_depth)
elif recursive and depth >= max_depth:
lines.append(f"{prefix}{indent}...")
else:
lines.append(f"{prefix}{connector}{entry}")
# Module: myagent.tools.search_files
import fnmatch
import os
import re
from typing import Any, Dict, List, Tuple
class SearchFilesTool(Tool):
@property
def name(self) -> str:
return "search_files"
@property
def description(self) -> str:
return "Search for a text pattern in files using substring or regex search."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Text or regex pattern to search for",
},
"path": {
"type": "string",
"description": "Directory or file to search (default: current directory)",
},
"regex": {
"type": "boolean",
"description": "Use regex matching (default: false = literal search)",
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to return (default: 50)",
},
"file_pattern": {
"type": "string",
"description": "Glob pattern for files to search (e.g., '*.py')",
},
},
"required": ["pattern"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
pattern = self._require(arguments, "pattern")
path = arguments.get("path", ".")
use_regex = arguments.get("regex", False)
max_results = arguments.get("max_results", 50)
file_pattern = arguments.get("file_pattern")
try:
if use_regex:
compiled = re.compile(pattern)
else:
compiled = None
target = os.path.abspath(path)
if os.path.isfile(target):
results = self._search_file(target, pattern, compiled, use_regex)
return self._format_results(results, max_results)
if os.path.isdir(target):
results = self._search_dir(target, pattern, compiled, use_regex, file_pattern, max_results)
return self._format_results(results, max_results)
return f"ERROR: Path not found: {target}"
except re.error as e:
return f"ERROR: Invalid regex pattern: {e}"
except Exception as e:
return f"ERROR: {e}"
def _search_file(self, file_path: str, pattern: str, compiled, use_regex: bool) -> List[Tuple[str, int, str]]:
results = []
if self._is_binary(file_path):
return results
try:
with open(file_path, "r", encoding="utf-8", errors="replace") as f:
for line_num, line in enumerate(f, start=1):
try:
if use_regex:
if compiled.search(line):
results.append((file_path, line_num, line.rstrip("\n")))
else:
if pattern in line:
results.append((file_path, line_num, line.rstrip("\n")))
except UnicodeDecodeError:
continue
except (OSError, PermissionError):
pass
return results
def _search_dir(self, root: str, pattern: str, compiled, use_regex: bool, file_pattern: str, max_results: int) -> List[Tuple[str, int, str]]:
results = []
exclude_dirs = {".git", ".hg", ".svn", "node_modules", "__pycache__", ".venv", "venv", ".claude", ".idea", ".vscode", "dist", "build"}
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [d for d in dirnames if d not in exclude_dirs]
for filename in filenames:
if file_pattern and not fnmatch.fnmatch(filename, file_pattern):
continue
file_path = os.path.join(dirpath, filename)
results.extend(self._search_file(file_path, pattern, compiled, use_regex))
if len(results) >= max_results:
return results[:max_results]
return results
def _format_results(self, results: List[Tuple[str, int, str]], max_results: int) -> str:
if not results:
return "No matches found."
lines = [f"Found {len(results)} matches:"]
for file_path, line_num, line in results[:max_results]:
display_line = line
if len(display_line) > 200:
display_line = display_line[:200] + " ... [truncated]"
lines.append(f" {file_path}:{line_num}: {display_line}")
if len(results) > max_results:
lines.append(f"\n... and {len(results) - max_results} more matches")
return "\n".join(lines)
def _is_binary(self, file_path: str) -> bool:
try:
with open(file_path, "rb") as f:
chunk = f.read(4096)
return b"\x00" in chunk
except:
return True
# Module: myagent.tools.edit_file
from typing import Any, Dict
class EditFileTool(Tool):
@property
def name(self) -> str:
return "edit_file"
@property
def description(self) -> str:
return "Replace text in an existing file. old_string must match exactly (including whitespace)."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file to edit",
},
"old_string": {
"type": "string",
"description": "Exact text to replace",
},
"new_string": {
"type": "string",
"description": "Replacement text",
},
},
"required": ["path", "old_string", "new_string"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
path = self._require(arguments, "path")
old_string = self._require(arguments, "old_string")
new_string = self._require(arguments, "new_string")
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
content = f.read()
if old_string not in content:
if old_string.strip() in content.strip():
return (
f"ERROR: old_string not found exactly in {path}. "
"Note: old_string must match exactly including surrounding whitespace."
)
return f"ERROR: old_string not found in {path}"
count = content.count(old_string)
if count > 1:
return (
f"ERROR: Found {count} occurrences of old_string in {path}. "
"Please make the old_string more specific (include more context)."
)
new_content = content.replace(old_string, new_string)
with open(path, "w", encoding="utf-8") as f:
f.write(new_content)
old_lines = old_string.count("\n")
new_lines = new_string.count("\n")
delta = new_lines - old_lines
return f"Successfully edited {path} (+{delta} lines)"
except FileNotFoundError:
return f"ERROR: File not found: {path}"
except PermissionError:
return f"ERROR: Permission denied: {path}"
except Exception as e:
return f"ERROR: {e}"
# Module: myagent.tools.write_file
import os
from typing import Any, Dict
class WriteFileTool(Tool):
@property
def name(self) -> str:
return "write_file"
@property
def description(self) -> str:
return "Write content to a file. Creates the file if it doesn't exist, overwrites if it does. Can optionally create parent directories."
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file to write",
},
"content": {
"type": "string",
"description": "Content to write to the file",
},
"create_dirs": {
"type": "boolean",
"description": "Create parent directories if they don't exist (default: true)",
},
},
"required": ["path", "content"],
}
def execute(self, arguments: Dict[str, Any]) -> str:
path = self._require(arguments, "path")
content = self._require(arguments, "content")
create_dirs = arguments.get("create_dirs", True)
try:
if create_dirs:
parent = os.path.dirname(os.path.abspath(path))
if parent and not os.path.exists(parent):
os.makedirs(parent, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(content)
lines = content.count("\n") + 1 if content else 0
return f"Successfully wrote {len(content)} characters, {lines} lines to {path}"
except PermissionError:
return f"ERROR: Permission denied: {path}"
except Exception as e:
return f"ERROR: {e}"
# Module: myagent.tools.registry
from typing import Any, Dict, List
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool) -> None:
if tool.name in self._tools:
raise ValueError(f"Tool '{tool.name}' is already registered")
self._tools[tool.name] = tool
def register_all(self, *tools: Tool) -> None:
for tool in tools:
self.register(tool)
def get(self, name: str) -> Tool:
if name not in self._tools:
raise KeyError(f"Tool '{name}' not found in registry")
return self._tools[name]
def execute(self, tool_name: str, arguments: Dict[str, Any]) -> str:
tool = self.get(tool_name)
try:
return tool.execute(arguments)
except Exception as e:
return f"ERROR: {type(e).__name__}: {e}"
def list_tools(self) -> List[str]:
return list(self._tools.keys())
def get_definitions(self) -> List[Dict[str, Any]]:
return [tool.to_definition() for tool in self._tools.values()]
def __contains__(self, name: str) -> bool:
return name in self._tools
# Module: myagent.agent
import json
import logging
import os
from typing import Any, Dict, List, Optional
logger = logging.getLogger("myagent.agent")
class Agent:
def __init__(
self,
llm: LLMProvider,
tool_registry: ToolRegistry,
permission: PermissionManager,
prompt_loader: PromptLoader,
max_turns: int = 50,
):
self.llm = llm
self.tools = tool_registry
self.permission = permission
self.prompt_loader = prompt_loader
self.max_turns = max_turns
system_prompt = self.prompt_loader.load()
logger.debug("System prompt length: %d chars", len(system_prompt))
self.context = ContextManager(system_prompt=system_prompt)
self.context.set_tool_definitions(self.tools.get_definitions())
self.turn_count = 0
logger.info("Agent initialized. max_turns=%d", max_turns)
def run(self, user_request: str) -> str:
logger.info(
"Starting task: %s%s",
user_request[:60],
"..." if len(user_request) > 60 else "",
)
print(f"\n{'='*60}")
print(f" MyAgent - Task: {user_request[:60]}{'...' if len(user_request) > 60 else ''}")
print(f"{'='*60}\n")
self.context.add_user_message(user_request)
final_response = ""
while self.turn_count < self.max_turns:
self.turn_count += 1
logger.info("--- Turn %d ---", self.turn_count)
print(f"\n--- Turn {self.turn_count} ---")
messages = self.context.build_messages()
tools = self.context.get_tool_definitions()
logger.debug("Sending %d messages to LLM", len(messages))
try:
response = self.llm.chat(messages, tools=tools)
except Exception as e:
logger.error("LLM error at turn %d: %s", self.turn_count, e, exc_info=True)
return f"LLM Error: {e}"
if response.has_tool_calls():
logger.info("Received %d tool call(s)", len(response.tool_calls))
history_tool_calls = []
for tc in response.tool_calls:
history_tool_calls.append({
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments),
},
})
self.context.add_assistant_message(
content=response.content,
tool_calls=history_tool_calls,
)
for tc in response.tool_calls:
arg_str = json.dumps(tc.arguments, ensure_ascii=False)
logger.info("Executing tool: %s(%s)", tc.name, arg_str)
print(f"\n[Tool] {tc.name}({arg_str})")
if tc.name not in self.tools:
result = f"ERROR: Unknown tool '{tc.name}'"
logger.warning("Unknown tool requested: %s", tc.name)
print(f"[Result] {result}")
self.context.add_tool_result(tc.id, tc.name, result)
continue
permitted = self.permission.check(tc.name, tc.arguments)
logger.debug("Permission check for %s: %s", tc.name, permitted)
if not permitted:
result = f"PERMISSION_DENIED: User declined to execute {tc.name}"
logger.warning("Permission denied for %s", tc.name)
print(f"[Result] {result}")
self.context.add_tool_result(tc.id, tc.name, result)
continue
result = self.tools.execute(tc.name, tc.arguments)
logger.info("Tool result: %s", result[:200] + ("..." if len(result) > 200 else ""))
print(f"[Result] {result[:500]}{'...' if len(result) > 500 else ''}")
self.context.add_tool_result(tc.id, tc.name, result)
else:
self.context.add_assistant_message(content=response.content)
final_response = response.content or "(no response)"
logger.info("Agent finished at turn %d", self.turn_count)
print(f"\n{'='*60}")
print(" Agent finished")
print(f"{'='*60}")
break
else:
final_response = f"Reached max turns ({self.max_turns}). Last response: {response.content or '(none)'}"
logger.warning("Reached max_turns limit (%s)", self.max_turns)
return final_response
def reset(self) -> None:
self.context.clear()
self.turn_count = 0
logger.info("Conversation reset")
def get_conversation_summary(self) -> str:
messages = self.context.get_messages()
user_msgs = sum(1 for m in messages if m.get("role") == "user")
assistant_msgs = sum(1 for m in messages if m.get("role") == "assistant")
tool_results = sum(1 for m in messages if m.get("role") == "tool")
summary = (
f"Conversation: {len(messages)} messages, "
f"{user_msgs} user, {assistant_msgs} assistant, {tool_results} tool results"
)
logger.debug("Summary: %s", summary)
return summary
# Module: myagent.main
import argparse
import getpass
import json
import os
import sys
from pathlib import Path
logger = get_logger(__name__)
def load_config(config_path: str) -> dict:
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
def resolve_api_key(llm_config: dict) -> str:
api_key = llm_config.get("api_key", "")
if api_key == "DEBUG":
logger.info("DEBUG mode: skipping API key resolution")
return api_key
if isinstance(api_key, str) and api_key.startswith("${") and api_key.endswith("}"):
env_var = api_key[2:-1]
api_key = os.environ.get(env_var, "")
logger.debug(
"Resolved API key from env var %s: %s",
env_var,
"set" if api_key else "empty",
)
if not api_key:
provider = llm_config.get("provider", "openai")
prompt_msg = f"API key not configured. Enter your API key for {provider}: "
api_key = getpass.getpass(prompt_msg)
llm_config["api_key"] = api_key
logger.info("API key entered via interactive prompt.")
return api_key
def setup_tools() -> ToolRegistry:
registry = ToolRegistry()
registry.register_all(
ReadFileTool(),
WriteFileTool(),
EditFileTool(),
ListFilesTool(),
SearchFilesTool(),
RunCommandTool(),
)
return registry
CONFIG_SKELETON = {
"llm": {
"provider": "sakura",
"api_key": "${API_KEY}",
"base_url": "https://api.ai.sakura.ad.jp/v1",
"model": "preview/Qwen3.6-35B-A3B",
"timeout": 120,
},
"permissions": {
"read_file": "auto",
"list_files": "auto",
"search_files": "auto",
"write_file": "ask",
"edit_file": "ask",
"run_command": "ask",
},
}
def init_config(config_path: str, force: bool = False) -> int:
path = Path(config_path)
if path.exists() and not force:
print(f"Config file already exists: {path}")
answer = input("Overwrite? [y/N]: ").strip().lower()
if answer not in ("y", "yes"):
print("Aborted.")
return 1
with open(path, "w", encoding="utf-8") as f:
json.dump(CONFIG_SKELETON, f, indent=2, ensure_ascii=False)
f.write("\n")
print(f"Created: {path}")
print("\nNext steps:")
print(" 1. Set your API key via environment variable (API_KEY)")
print(" 2. Or edit the file directly to set 'api_key'")
print(f" 3. Run: python -m myagent.main -i")
return 0
def main():
parser = argparse.ArgumentParser(
description="MyAgent - Autonomous Coding Agent",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"request",
nargs="?",
help="Task to perform (e.g., 'create hello.py')",
)
parser.add_argument(
"--config",
default="config.json",
help="Path to config file (default: config.json)",
)
parser.add_argument(
"--prompts",
default="prompts",
help="Path to prompts directory (default: prompts)",
)
parser.add_argument(
"--max-turns",
type=int,
default=50,
help="Maximum agent turns (default: 50)",
)
parser.add_argument(
"--interactive",
"-i",
action="store_true",
help="Run in interactive mode",
)
parser.add_argument(
"--log-dir",
default="logs",
help="Directory for log files (default: logs)",
)
parser.add_argument(
"--init",
action="store_true",
help="Create a skeleton config.json and exit",
)
parser.add_argument(
"--force",
"-f",
action="store_true",
help="Overwrite existing config when using --init",
)
args = parser.parse_args()
if args.init:
return init_config(args.config, force=args.force)
log_file = setup_logging(log_dir=args.log_dir)
config_path = os.path.abspath(args.config)
if not os.path.exists(config_path):
logger.error("Config file not found: %s", config_path)
print(f"Error: Config file not found: {config_path}")
print("Create a config.json with your API settings.")
sys.exit(1)
config = load_config(config_path)
llm_config = config.get("llm", {})
resolve_api_key(llm_config)
logger.debug("Initializing LLM provider")
llm = create_provider(llm_config)
logger.debug("Registering tools")
tools = setup_tools()
perm_overrides = config.get("permissions", {})
permission = PermissionManager(perm_overrides)
logger.debug("Loading prompts from: %s", args.prompts)
prompt_loader = PromptLoader(args.prompts)
logger.debug("Creating agent with max_turns=%d", args.max_turns)
agent = Agent(
llm=llm,
tool_registry=tools,
permission=permission,
prompt_loader=prompt_loader,
max_turns=args.max_turns,
)
print(f"System prompt loaded from: {args.prompts}/")
print(f"Log file: {log_file}")
print(f"Available tools: {', '.join(tools.list_tools())}")
print(permission.get_summary())
print()
if args.interactive or not args.request:
print("MyAgent Interactive Mode")
print("Type 'exit' or 'quit' to exit, 'reset' to clear history.\n")
while True:
try:
user_input = input(">>> ").strip()
except (EOFError, KeyboardInterrupt):
print("\nExiting...")
break
if not user_input:
continue
if user_input.lower() in ("exit", "quit"):
print("Exiting...")
break
if user_input.lower() == "reset":
agent.reset()
print("Conversation reset.")
continue
if user_input.lower() == "summary":
print(agent.get_conversation_summary())
continue
result = agent.run(user_input)
print(f"\n{result}\n")
else:
result = agent.run(args.request)
print(result)
if __name__ == "__main__":
main()