#!/usr/bin/env python3 """Package doc_redaction LangGraph agent code into an AgentCore app folder.""" from __future__ import annotations import argparse import os import re import shutil import stat import subprocess from pathlib import Path import tomllib _COPY_IGNORE = shutil.ignore_patterns( "__pycache__", "*.pyc", ".pytest_cache", ".mypy_cache" ) # Runtime Python deps to merge into the AgentCore app's pyproject.toml (not full pi-agent stack). RUNTIME_DEPENDENCIES: dict[str, str] = { "gradio_client": ">=1.0.0", "httpx": ">=0.28.0", "python-dotenv": ">=1.0.0", "langchain-openai": ">=1.0.0", "langchain-core": ">=1.0.0", "langgraph": ">=1.0.2", "langchain-aws": ">=1.0.0", "pymupdf": ">=1.24.0", "pandas": ">=2.0.0", } MAIN_PY = '''"""doc_redaction LangGraph agent — packaged by agent-redact/agentcore/package_runtime.py.""" from __future__ import annotations import sys from pathlib import Path _APP_ROOT = Path(__file__).resolve().parent _PI_DIR = _APP_ROOT / "pi" for path in (_APP_ROOT, _PI_DIR): text = str(path) if text not in sys.path: sys.path.insert(0, text) from invoke_agent import bootstrap_runtime_env, invoke_redaction_agent # noqa: E402 bootstrap_runtime_env(_APP_ROOT) from bedrock_agentcore import BedrockAgentCoreApp # noqa: E402 app = BedrockAgentCoreApp() @app.entrypoint async def handler(request: dict): async for event in invoke_redaction_agent(request): yield event if __name__ == "__main__": app.run() ''' def _repo_root() -> Path: return Path(__file__).resolve().parents[2] def _default_agentcore_app() -> Path: return _repo_root() / "agent-redact" / "RedactionAgent" / "app" / "RedactionAgent" def _rmtree_robust(path: Path) -> None: """Remove a directory tree on Windows / OneDrive (clears read-only files first).""" def _on_rm_error(func, location, _exc_info) -> None: os.chmod(location, stat.S_IWRITE) func(location) shutil.rmtree(path, onerror=_on_rm_error) def _copy_tree(src: Path, dest: Path, *, dry_run: bool) -> None: if dry_run: print(f" copy tree {src} -> {dest}") return if dest.exists(): _rmtree_robust(dest) shutil.copytree(src, dest, ignore=_COPY_IGNORE) def _copy_file(src: Path, dest: Path, *, dry_run: bool) -> None: if dry_run: print(f" copy file {src} -> {dest}") return dest.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dest) def _replace_dependencies_block(text: str, deps: list[str]) -> str: lines = text.splitlines() out: list[str] = [] index = 0 while index < len(lines): if lines[index].strip().startswith("dependencies"): out.append("dependencies = [") for dep in deps: out.append(f' "{dep}",') out.append("]") index += 1 while index < len(lines) and lines[index].strip() != "]": index += 1 index += 1 continue out.append(lines[index]) index += 1 return "\n".join(out) + ("\n" if text.endswith("\n") else "") def _merge_pyproject(pyproject_path: Path, *, dry_run: bool) -> None: text = pyproject_path.read_text(encoding="utf-8") if dry_run: print(f" merge deps into {pyproject_path}") return try: data = tomllib.loads(text) except tomllib.TOMLDecodeError as exc: raise SystemExit(f"Could not parse {pyproject_path}: {exc}") from exc existing: dict[str, str] = {} for item in data.get("project", {}).get("dependencies", []): if isinstance(item, str): name = re.split(r"[<>=!~\[]", item, maxsplit=1)[0].strip() existing[name.lower()] = item for name, spec in RUNTIME_DEPENDENCIES.items(): key = name.lower() if key not in existing: existing[key] = f"{name}{spec}" merged = [existing[k] for k in sorted(existing, key=str.lower)] pyproject_path.write_text( _replace_dependencies_block(text, merged), encoding="utf-8" ) def package_runtime( target: Path, *, dry_run: bool = False, ) -> list[str]: """Sync monorepo redaction agent sources into *target* (AgentCore app folder).""" repo = _repo_root() agent_redact = repo / "agent-redact" agentcore = agent_redact / "agentcore" actions: list[str] = [] def log(msg: str) -> None: actions.append(msg) print(msg) log(f"Packaging doc_redaction runtime -> {target}") _copy_tree( agent_redact / "redaction_langgraph", target / "redaction_langgraph", dry_run=dry_run, ) pi_dest = target / "pi" for name in ("remote_redaction.py",): _copy_file(agent_redact / "pi" / name, pi_dest / name, dry_run=dry_run) _copy_file( agentcore / "bundle_support" / "session_workspace.py", pi_dest / "session_workspace.py", dry_run=dry_run, ) for module in ("invoke_agent.py", "session_store.py", "workspace_sync.py"): _copy_file(agentcore / module, target / module, dry_run=dry_run) if dry_run: log(f" write {target / 'main.py'}") else: (target / "main.py").write_text(MAIN_PY, encoding="utf-8") log(f"wrote {target / 'main.py'}") pyproject = target / "pyproject.toml" if pyproject.is_file(): _merge_pyproject(pyproject, dry_run=dry_run) log(f"merged runtime dependencies into {pyproject}") elif dry_run: log(f" skip pyproject merge (no {pyproject} — run agentcore create first)") else: raise SystemExit(f"Missing {pyproject} — run agentcore create first.") env_example = target / "agentcore.env.example" env_local = target / "agentcore.env" example_text = """# Loaded at runtime startup when present in the CodeZip (see invoke_agent.bootstrap_runtime_env). # Also set these on the AgentCore runtime in AWS if you prefer console/config-bundle env. # CDK + AgentCore: use main Express HTTPS (ExpressServiceEndpoint), not Service Connect. DOC_REDACTION_GRADIO_URL=https://your-doc-redaction-host.example AGENT_DEFAULT_PROVIDER=amazon-bedrock AGENT_DEFAULT_MODEL=anthropic.claude-sonnet-4-6 AWS_REGION=eu-west-2 AGENT_WORKSPACE_DIR=/tmp/agentcore-workspace AGENT_DEFAULT_OCR_METHOD=paddle AGENT_DEFAULT_PII_METHOD=Local """ if dry_run: log(f" write {env_example}") if env_local.is_file(): log(f" keep existing {env_local}") else: env_example.write_text(example_text, encoding="utf-8") log(f"wrote {env_example}") if not env_local.is_file(): env_local.write_text(example_text, encoding="utf-8") log(f"wrote {env_local} (copy from example — edit before deploy)") return actions def run_deploy(agentcore_project: Path) -> None: env = dict(**{k: v for k, v in __import__("os").environ.items()}) env.setdefault("UV_LINK_MODE", "copy") subprocess.run( ["agentcore", "deploy"], cwd=str(agentcore_project), check=True, env=env, ) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Package doc_redaction LangGraph agent into an AgentCore app folder.", ) parser.add_argument( "--target", type=Path, default=_default_agentcore_app(), help="AgentCore app folder (default: agent-redact/RedactionAgent/app/RedactionAgent)", ) parser.add_argument( "--dry-run", action="store_true", help="Print actions without writing files", ) parser.add_argument( "--deploy", action="store_true", help="Run agentcore deploy from the RedactionAgent project after packaging", ) args = parser.parse_args(argv) target = args.target.resolve() package_runtime(target, dry_run=args.dry_run) if args.deploy: if args.dry_run: print("Skipping deploy (--dry-run).") return 0 project = target.parent.parent if not (project / "agentcore" / "agentcore.json").is_file(): raise SystemExit(f"Not an AgentCore project: {project}") print(f"Running agentcore deploy in {project} ...") run_deploy(project) print("Done.") return 0 if __name__ == "__main__": raise SystemExit(main())