エージェント × Python── Copilot Studio と Agents SDK に Python の処理を足す

"""agent_routes.py: command routes, a ranked catch-all and an error handler for an AgentApplication."""
import re

from microsoft_agents.hosting.core import AgentApplication, RouteRank, TurnContext, TurnState

HELP = "Commands: /help, status <request id> (for example: status R-2026-002)."
STATUS_RE = re.compile(r"^status\s+(R-\d{4}-\d{3})$", re.IGNORECASE)


async def on_help(context: TurnContext, _state: TurnState):
    await context.send_activity(HELP)


async def on_status(context: TurnContext, _state: TurnState):
    request_id = STATUS_RE.match(context.activity.text.strip()).group(1).upper()
    await context.send_activity(f"Looking up {request_id} ...")


async def on_other(context: TurnContext, _state: TurnState):
    await context.send_activity("I did not recognise that. Type /help for commands.")


async def on_error(context: TurnContext, error: Exception):
    print(f"turn error: {type(error).__name__}")  # write details to your log, not to the chat
    await context.send_activity("Something went wrong. Please try again later.")


def register_routes(app: AgentApplication) -> AgentApplication:
    app.conversation_update("membersAdded")(on_help)
    app.message("/help")(on_help)
    app.message(STATUS_RE)(on_status)
    app.activity("message", rank=RouteRank.LAST)(on_other)  # catch-all, evaluated after the routes above
    app.error(on_error)
    return app

"""app.py: build an Agents SDK agent from environment variables and serve it with aiohttp."""
import os

from aiohttp.web import Application, Request, Response, run_app
from microsoft_agents.activity import load_configuration_from_env
from microsoft_agents.authentication.msal import MsalConnectionManager
from microsoft_agents.hosting.aiohttp import CloudAdapter, jwt_authorization_middleware, start_agent_process
from microsoft_agents.hosting.core import AgentApplication, Authorization, MemoryStorage, TurnContext, TurnState

PREFIX = "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__"


def check_settings(env):
    """Return problems that should stop start-up."""
    keys = {k[len(PREFIX):] for k in env if k.startswith(PREFIX)}
    anonymous = str(env.get(PREFIX + "ANONYMOUS_ALLOWED", "")).lower() == "true"
    problems = []
    if not keys:
        problems.append("no SERVICE_CONNECTION settings")
    if anonymous and env.get("AGENT_HOST", "localhost") not in ("localhost", "127.0.0.1"):
        problems.append("ANONYMOUS_ALLOWED is for local development only")
    if not anonymous and keys and "CLIENTID" not in keys and env.get(PREFIX + "AUTHTYPE") != "SystemManagedIdentity":
        problems.append("CLIENTID is missing")
    if "CLIENTSECRET" in keys:
        problems.append("client secret configured: use Certificate or a managed identity instead")
    return problems


def build_agent(env):
    config = load_configuration_from_env(env)
    storage = MemoryStorage()
    connections = MsalConnectionManager(**config)
    agent = AgentApplication[TurnState](
        storage=storage,
        adapter=CloudAdapter(connection_manager=connections),
        authorization=Authorization(storage, connections, **config),
        **config,
    )

    @agent.activity("message")
    async def on_message(context: TurnContext, _state: TurnState):
        await context.send_activity(f"you said: {context.activity.text}")

    return agent, connections


def build_web_app(agent, connections):
    async def entry_point(req: Request) -> Response:
        return await start_agent_process(req, agent, agent.adapter)

    web = Application(middlewares=[jwt_authorization_middleware])
    web.router.add_post("/api/messages", entry_point)
    web["agent_configuration"] = connections.get_default_connection_configuration()
    web["agent_app"] = agent
    web["adapter"] = agent.adapter
    return web


if __name__ == "__main__":
    issues = check_settings(os.environ)
    if issues:
        raise SystemExit("stopped: " + "; ".join(issues))
    agent, connections = build_agent(os.environ)
    run_app(build_web_app(agent, connections), host=os.environ.get("AGENT_HOST", "localhost"),
            port=int(os.environ.get("PORT", 3978)))

"""function_app.py: HTTP-triggered lookup that an agent or a Copilot Studio REST API tool calls."""
import csv
import json
import logging
import os

import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
COLUMNS = {"文書名": "title", "種別": "doc_type", "状態": "state", "指摘件数": "findings", "完了日": "completed"}


def load_ledger(path):
    with open(path, encoding="utf-8-sig", newline="") as f:
        return {row["依頼ID"]: row for row in csv.DictReader(f)}


def json_response(body, status_code=200):
    return func.HttpResponse(json.dumps(body, ensure_ascii=False), status_code=status_code,
                             mimetype="application/json")


@app.route(route="request_status", methods=["GET"])
def request_status(req: func.HttpRequest) -> func.HttpResponse:
    request_id = (req.params.get("id") or "").strip().upper()
    if not request_id:
        return json_response({"error": "query parameter 'id' is required"}, 400)
    row = load_ledger(os.getenv("LEDGER_PATH", "ledger.csv")).get(request_id)
    if row is None:
        return json_response({"id": request_id, "found": False})
    logging.info("request_status looked up %s", request_id)
    result = {"id": request_id, "found": True}
    result.update({en: row[ja] for ja, en in COLUMNS.items()})
    result["findings"] = int(result["findings"] or 0)
    return json_response(result)

"""Unit-test a Python v2 HTTP function without Core Tools (function and tests in one file for brevity)."""
import json
import sys
import unittest

import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)


@app.route(route="count_findings", methods=["POST"])
def count_findings(req: func.HttpRequest) -> func.HttpResponse:
    try:
        rows = req.get_json().get("rows", [])
    except ValueError:
        return func.HttpResponse("JSON body is required", status_code=400)
    total = sum(int(r.get("findings", 0)) for r in rows)
    return func.HttpResponse(json.dumps({"rows": len(rows), "findings": total}), mimetype="application/json")


def make_request(body: bytes) -> func.HttpRequest:
    return func.HttpRequest(method="POST", url="/api/count_findings", body=body)


class CountFindingsTest(unittest.TestCase):
    def call(self, body):
        return count_findings.build().get_user_function()(make_request(body))

    def test_sum(self):
        resp = self.call(json.dumps({"rows": [{"findings": 2}, {"findings": "3"}]}).encode())
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(json.loads(resp.get_body()), {"rows": 2, "findings": 5})

    def test_missing_rows(self):
        self.assertEqual(json.loads(self.call(b"{}").get_body()), {"rows": 0, "findings": 0})

    def test_bad_json(self):
        self.assertEqual(self.call(b"not json").status_code, 400)

    def test_direct_call(self):
        resp = count_findings(make_request(json.dumps({"rows": [{"findings": 4}]}).encode()))
        self.assertEqual(json.loads(resp.get_body())["findings"], 4)


if __name__ == "__main__":
    suite = unittest.defaultTestLoader.loadTestsFromTestCase(CountFindingsTest)
    result = unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
    print("tests passed" if result.wasSuccessful() else "tests failed")
    sys.exit(0 if result.wasSuccessful() else 1)

"""Write an OpenAPI 2.0 (swagger) file for the request_status function, for a Copilot Studio REST API tool."""
import json
import os

HOST = os.environ.get("FUNCTION_HOST", "localhost:7071")  # set to your function app's host name

SPEC = {
    "swagger": "2.0",
    "info": {
        "title": "Request status",
        "version": "1.0.0",
        "description": "Looks up a document review request by its ID (for example R-2026-002). Use it when a user "
                       "asks about the state, number of findings or completion date of a request.",
    },
    "host": HOST,
    "basePath": "/api",
    "schemes": ["http" if HOST.startswith("localhost") else "https"],
    "produces": ["application/json"],
    "securityDefinitions": {"functionKey": {"type": "apiKey", "name": "x-functions-key", "in": "header"}},
    "security": [{"functionKey": []}],
    "paths": {
        "/request_status": {
            "get": {
                "operationId": "GetRequestStatus",
                "summary": "Get request status",
                "description": "Returns title, document type, state, number of findings and completion date "
                               "for one request ID. Returns found=false when the ID is not in the ledger.",
                "parameters": [{"name": "id", "in": "query", "required": True, "type": "string",
                                "description": "Request ID such as R-2026-002"}],
                "responses": {
                    "200": {"description": "Lookup result", "schema": {"type": "object", "properties": {
                        "id": {"type": "string", "description": "Request ID"},
                        "found": {"type": "boolean", "description": "False when the ID is unknown"},
                        "title": {"type": "string", "description": "Document title"},
                        "doc_type": {"type": "string", "description": "Document type"},
                        "state": {"type": "string", "description": "Current review state"},
                        "findings": {"type": "integer", "description": "Number of findings"},
                        "completed": {"type": "string", "description": "Completion date, empty if open"}}}},
                    "400": {"description": "The id parameter is missing"},
                },
            }
        }
    },
}


def check(spec):
    problems = []
    if spec.get("swagger") != "2.0":
        problems.append("REST API tools need an OpenAPI v2 (swagger 2.0) file")
    if len(spec["info"].get("description", "")) < 40:
        problems.append("API description is too short for tool selection")
    for path, ops in spec["paths"].items():
        for method, op in ops.items():
            where = f"{method.upper()} {path}"
            if len(op.get("description", "")) < 40:
                problems.append(f"{where}: operation description is too short")
            for p in op.get("parameters", []):
                if not p.get("description"):
                    problems.append(f"{where}: parameter {p['name']} has no description")
            for code, resp in op.get("responses", {}).items():
                for name, prop in resp.get("schema", {}).get("properties", {}).items():
                    if not prop.get("description"):
                        problems.append(f"{where}: response {code} field {name} has no description")
    return problems


if __name__ == "__main__":
    issues = check(SPEC)
    with open("out_openapi.json", "w", encoding="utf-8") as f:
        json.dump(SPEC, f, ensure_ascii=False, indent=2)
    print("wrote out_openapi.json; operations:", sum(len(v) for v in SPEC["paths"].values()))
    print("issues:", issues or "none")

"""Agent handler that calls the request_status function and replies with the result."""
import os
import re

from aiohttp import ClientError, ClientSession, ClientTimeout
from microsoft_agents.hosting.core import TurnContext, TurnState

STATUS_RE = re.compile(r"^status\s+(R-\d{4}-\d{3})$", re.IGNORECASE)
TIMEOUT = ClientTimeout(total=float(os.environ.get("STATUS_API_TIMEOUT", "20")))


async def fetch_status(request_id):
    url = os.environ.get("STATUS_API_URL", "http://localhost:7071/api/request_status")
    headers = {}
    function_key = os.environ.get("STATUS_API_KEY")  # from app settings or Key Vault, never from code
    if function_key:
        headers["x-functions-key"] = function_key
    async with ClientSession(timeout=TIMEOUT) as session:
        async with session.get(url, params={"id": request_id}, headers=headers) as resp:
            resp.raise_for_status()
            return await resp.json()


def format_reply(data):
    if not data.get("found"):
        return f"{data.get('id')}: not found in the ledger."
    done = data.get("completed") or "-"
    return (f"{data['id']} {data['title']} ({data['doc_type']}): {data['state']}, "
            f"findings {data['findings']}, completed {done}")


async def on_status(context: TurnContext, _state: TurnState):
    request_id = STATUS_RE.match(context.activity.text.strip()).group(1).upper()
    try:
        data = await fetch_status(request_id)
    except (ClientError, TimeoutError) as error:
        print(f"status lookup failed: {type(error).__name__}")
        await context.send_activity("The status service did not respond. Please try again later.")
        return
    await context.send_activity(format_reply(data))

# Register it on your AgentApplication, for example: AGENT_APP.message(STATUS_RE)(on_status)

"""Ask a published Copilot Studio agent one question from Python (signed-in user)."""
import asyncio
import os

from msal import PublicClientApplication
from microsoft_agents.activity import ActivityTypes
from microsoft_agents.copilotstudio.client import ConnectionSettings, CopilotClient

SCOPES = ["https://api.powerplatform.com/.default"]
NAMES = ["ENVIRONMENTID", "SCHEMANAME", "AGENTAPPID", "TENANTID"]


def settings_from_env():
    values = {n: os.environ.get("COPILOTSTUDIOAGENT__" + n, "") for n in NAMES}
    missing = [n for n, v in values.items() if not v]
    if missing:
        raise SystemExit("set COPILOTSTUDIOAGENT__" + ", COPILOTSTUDIOAGENT__".join(missing))
    return values


def acquire_user_token(app_id, tenant_id):
    app = PublicClientApplication(client_id=app_id, authority=f"https://login.microsoftonline.com/{tenant_id}")
    accounts = app.get_accounts()
    result = app.acquire_token_silent(SCOPES, account=accounts[0]) if accounts else None
    if not result:
        result = app.acquire_token_interactive(scopes=SCOPES)
    return result["access_token"]


async def ask_once(question):
    v = settings_from_env()
    settings = ConnectionSettings(
        environment_id=v["ENVIRONMENTID"],
        agent_identifier=v["SCHEMANAME"],
        cloud=None,
        copilot_agent_type=None,
        custom_power_platform_cloud=None,
    )
    client = CopilotClient(settings, acquire_user_token(v["AGENTAPPID"], v["TENANTID"]))
    conversation_id = None
    async for activity in client.start_conversation(True):
        conversation_id = activity.conversation.id
    replies = []
    async for reply in client.ask_question(question, conversation_id):
        if reply.type == ActivityTypes.message and reply.text:
            replies.append(reply.text)
    return replies


if __name__ == "__main__":
    for text in asyncio.run(ask_once(os.environ.get("QUESTION", "What can you help me with?"))):
        print(text)